365 Python Projects for DevOps Engineers Page 1

365 Python Projects

for DevOps Engineers
From Absolute Beginner to Production-Grade DevOps Automation

Every script explained line-by-line — ready to push to

GitHub/GitLab

365 8 3 ∞
Projects Learning Parts Difficulty Tiers Real-World Uses

■
■ Linux ■ Docker Kubernetes ■ AWS ■ CI/CD ■ Monitoring

■ Terraform ■ Ansible ■ Git ■ Bash ■ Networking ■ Security

By Akash Francis | Maintenance Engineer → Cloud/DevOps Engineer

[Link]/akashfrancis3211 | [Link]/in/akash-francis91
Edition: 2025 | Generated: April 16, 2026

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 2

How to Use This Book

■ Structure
This book contains 365 Python projects organised into 8 thematic parts, progressing from absolute Python
basics through to production-grade DevOps automation. Each project is a complete, runnable script with
every line explained.

■ Line-by-Line Explanations
After each code block you will find a breakdown of every significant line — what it does, why it is written that
way, and what happens if you omit it. This is not just 'what' but 'why'.

■ GitHub/GitLab Ready
Each project is a self-contained .py file. The companion repository structure
(part_01_foundations/day_001_hello_world.py) mirrors the book chapters. Clone the repo, run the script,
read the explanation — learn by doing.

■ Prerequisites
Python 3.8 or newer. A Linux terminal (Ubuntu/Debian recommended). Docker for container projects. AWS
CLI configured for cloud projects. All third-party libraries are noted at the top of each script with pip install
instructions.

■ Learning Path
Days 1-30: Python core + CLI scripting. Days 31-80: File I/O, YAML, JSON, automation. Days 81-130:
Networking, HTTP, APIs. Days 131-180: Docker and containers. Days 181-230: AWS and cloud automation.
Days 231-280: CI/CD pipelines and testing. Days 281-330: Kubernetes and infrastructure. Days 331-365:
Monitoring, observability, and capstone projects.

Difficulty Legend

Beginner Core Python, standard library only. No prior DevOps knowledge required.

Intermediate Common third-party libraries. Assumes Linux comfort and basic DevOps concepts.

Advanced Production patterns, cloud SDKs, Kubernetes APIs. Builds on all prior projects.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 3

Table of Contents
PART I — PYTHON FOUNDATIONS (Days 1–30)
Day 1: Hello, DevOps World — Beginner
Day 2: Environment Variable Manager — Beginner
Day 3: File & Directory Scanner — Beginner
Day 4: Log File Parser — Beginner
Day 5: System Health Monitor — Beginner

PART II — FILES, DATA & AUTOMATION (Days 31–80)

Day 31: YAML Config Reader & Validator — Intermediate
Day 32: JSON to CSV Converter — Beginner
Day 33: Automated Backup Script — Intermediate

PART III — NETWORKING & APIs (Days 81–130)

Day 81: HTTP Health Checker — Intermediate
Day 82: REST API Client with Retry — Intermediate

PART IV — DOCKER & CONTAINERS (Days 131–180)

Day 131: Docker Container Inspector — Intermediate
Day 132: Dockerfile Linter — Intermediate

PART V — AWS & CLOUD (Days 181–230)

Day 181: AWS S3 Bucket Manager — Intermediate
Day 182: EC2 Instance Manager — Intermediate

PART VI — CI/CD & AUTOMATION (Days 231–280)

Day 231: GitHub Actions Workflow Validator — Advanced

PART VII — KUBERNETES & INFRASTRUCTURE (Days 281–330)

Day 281: Kubernetes Manifest Validator — Advanced

PART VIII — MONITORING & OBSERVABILITY (Days 331–365)

Day 331: Prometheus Metrics Exporter — Advanced
Day 365: Full DevOps Pipeline Orchestrator — Advanced

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 4

PART I — PYTHON FOUNDATIONS

(Days 1–30)

Each project in this part is a complete, runnable Python script. Read the description,

5 study the code, then check the line-by-line explanation below it. Type the code
yourself for best retention.
Projects

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 5

DAY 001 BEGINNER

Hello, DevOps World

Your first Python script — printing system info to stdout, the DevOps way.

■ Source Code
1 import platform # Standard library for OS/hardware info
2 import sys # Provides access to Python interpreter details
3 import os # Operating system interface
4
5 def main():
6 # [Link]() returns 'Linux', 'Windows', or 'Darwin'
7 print(f"OS : {[Link]()} {[Link]()}")
8 # [Link] gives the full Python version string
9 print(f"Python : {[Link]()[0]}")
10 # [Link]() returns the current working directory
11 print(f"Working Dir : {[Link]()}")
12 # [Link]() returns the login name of the current user
13 try:
14 print(f"User : {[Link]()}")
15 except OSError:
16 # Fallback when running in non-interactive shells (CI/CD)
17 print(f"User : {[Link]('USER', 'unknown')}")
18
19 if __name__ == "__main__":
20 # This guard ensures main() only runs when script is executed directly
21 # not when it is imported as a module
22 main()

■ Line-by-Line Explanation

1 import platform
loads the platform module from Python's standard library.

2 [Link]()
returns the OS name; essential for writing cross-platform DevOps scripts.

3 [Link]
a string like '3.11.0 (default, ...)'; we split on whitespace and take index 0.

4 [Link]()
Current Working Directory; important to know when scripts run in pipelines.

5 try/except OSError
[Link]() can fail inside Docker containers or CI runners. Always handle it.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 6

6 if __name__ == '__main__'
Python sets __name__ to '__main__' only when the file is executed directly. This pattern is the foundation of
reusable, importable modules.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 7

DAY 002 BEGINNER

Environment Variable Manager

Read, set, and validate environment variables — core to 12-factor app config.

■ Source Code
1 import os
2 import sys
3
4 # Dictionary of required environment variables and their descriptions
5 REQUIRED_VARS = {
6 "DB_HOST": "Database hostname",
7 "DB_PORT": "Database port (e.g. 5432)",
8 "APP_ENV": "Deployment environment (dev/staging/prod)",
9 }
10
11 def check_env():
12 """Check that all required environment variables are set."""
13 missing = [] # List to collect missing variable names
14
15 for var, description in REQUIRED_VARS.items():
16 # [Link]() returns None if variable is not set
17 # unlike [Link][var] which raises KeyError
18 value = [Link](var)
19 if value is None:
20 [Link](f" - {var}: {description}")
21 else:
22 # Mask secrets: only show first 3 chars + asterisks
23 masked = value[:3] + "*" * (len(value) - 3) if len(value) > 3 else "***"
24 print(f"[OK] {var} = {masked}")
25
26 if missing:
27 print("\n[ERROR] Missing required environment variables:")
28 for m in missing:
29 print(m)
30 # Exit with code 1 to signal failure to the calling process/pipeline
31 [Link](1)
32 else:
33 print("\n[PASS] All required environment variables are set.")
34
35 def set_defaults():
36 """Set safe default values for optional variables."""
37 defaults = {
38 "LOG_LEVEL": "INFO",
39 "MAX_RETRIES": "3",
40 "TIMEOUT": "30",
41 }

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 8

42 for key, value in [Link]():

43 # [Link]() only sets if the key doesn't already exist
44 # This respects values already set by the shell or CI system
45 [Link](key, value)
46 print(f"[DEFAULT] {key} = {[Link][key]}")
47
48 if __name__ == "__main__":
49 set_defaults()
50 check_env()

■ Line-by-Line Explanation

1 REQUIRED_VARS dict
Maps variable names to human-readable descriptions. This pattern makes your validation self-documenting.

2 [Link](var)
Safe dictionary-style access. Returns None instead of raising an exception if the key is missing.

3 [Link](1)
Exits the process with a non-zero return code. CI/CD pipelines (Jenkins, GitHub Actions) treat any non-zero exit as
a FAILURE and stop the pipeline.

4 Masking secrets
Never print raw secrets to logs. The slice value[:3] + '*'*(len-3) pattern reveals enough to confirm the right variable
is set without exposing the value.

5 [Link]()
Atomic check-and-set: if the key exists, it does nothing; if not, it sets the default. Safe to call multiple times.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 9

DAY 003 BEGINNER

File & Directory Scanner

Recursively walk a directory tree, categorize files by extension, and report sizes.

■ Source Code
1 import os
2 import sys
3 from collections import defaultdict # Dict that auto-initialises missing keys
4
5 def human_readable(size_bytes: int) -> str:
6 """Convert bytes to a human-readable string (KB, MB, GB)."""
7 # Iterate through units; each step is 1024x larger
8 for unit in ["B", "KB", "MB", "GB", "TB"]:
9 if size_bytes < 1024.0:
10 return f"{size_bytes:.1f} {unit}"
11 size_bytes /= 1024.0
12 return f"{size_bytes:.1f} PB"
13
14 def scan_directory(root_path: str) -> dict:
15 """Walk the directory tree and collect stats per file extension."""
16 stats = defaultdict(lambda: {"count": 0, "total_size": 0})
17
18 # [Link]() is a generator that yields (dirpath, dirnames, filenames)
19 # It traverses the entire tree depth-first by default
20 for dirpath, dirnames, filenames in [Link](root_path):
21 # Skip hidden directories (like .git) to avoid noise
22 # Modifying dirnames IN-PLACE prunes the walk — [Link] won't descend
23 dirnames[:] = [d for d in dirnames if not [Link](".")]
24
25 for filename in filenames:
26 # [Link] builds the full path correctly on any OS
27 full_path = [Link](dirpath, filename)
28 # [Link] splits "[Link]" into ("[Link]", ".gz")
29 _, ext = [Link](filename)
30 ext = [Link]() if ext else "(no extension)"
31
32 try:
33 # [Link] returns file size in bytes
34 size = [Link](full_path)
35 stats[ext]["count"] += 1
36 stats[ext]["total_size"] += size
37 except (PermissionError, OSError):
38 # Files might be unreadable (sockets, broken symlinks)
39 pass
40
41 return dict(stats)

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 10

42
43 def report(stats: dict, root_path: str):
44 """Print a formatted summary table."""
45 print(f"\nDirectory Scan Report: {root_path}")
46 print("-" * 50)
47 print(f"{'Extension':<20} {'Files':>6} {'Total Size':>12}")
48 print("-" * 50)
49
50 # Sort by total size descending to see largest categories first
51 for ext, data in sorted([Link](), key=lambda x: -x[1]["total_size"]):
52 print(f"{ext:<20} {data['count']:>6} {human_readable(data['total_size']):>12}")
53
54 if __name__ == "__main__":
55 path = [Link][1] if len([Link]) > 1 else "."
56 if not [Link](path):
57 print(f"Error: '{path}' is not a directory.")
58 [Link](1)
59 stats = scan_directory(path)
60 report(stats, path)

■ Line-by-Line Explanation

1 defaultdict(lambda: {...})
A dict that auto-creates a new {'count':0,'total_size':0} entry for any unseen key. Avoids key-not-found errors.

2 [Link]() generator
Memory-efficient recursive traversal. It yields one tuple per directory without loading the whole tree into memory.

3 dirnames[:] = [...]
Modifying the list IN-PLACE (slice assignment) is the only way to prune [Link]'s traversal. Assigning a new list
(dirnames = [...]) has no effect.

4 [Link]
Always use this instead of splitting on '.' yourself. It correctly handles files like '.bashrc' (no extension) and
'[Link]'.

5 [Link][1]
argv[0] is the script name; argv[1] is the first argument. Defaulting to '.' means the script works with no arguments.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 11

DAY 004 BEGINNER

Log File Parser

Parse an NGINX/Apache log file, count status codes, and find top IPs.

■ Source Code
1 import re # Regular expressions
2 import sys
3 from collections import Counter # Efficient frequency counting
4
5 # NGINX combined log format pattern
6 # Each named group (?P<name>...) creates a key in the match dictionary
7 LOG_PATTERN = [Link](
8 r'(?P<ip>\S+)\s+' # Client IP address
9 r'\S+\s+\S+\s+' # ident and auth fields (usually -)
10 r'\[(?P<time>[^\]]+)\]\s+' # Timestamp inside square brackets
11 r'"(?P<method>\S+)\s+' # HTTP method (GET, POST, etc.)
12 r'(?P<path>\S+)\s+\S+"\s+'# Request path + HTTP version
13 r'(?P<status>\d{3})\s+' # 3-digit HTTP status code
14 r'(?P<size>\d+|-)' # Response size in bytes (or - if 0)
15 )
16
17 def parse_log(filepath: str):
18 """Parse a log file and return lists of parsed entries."""
19 status_counts = Counter() # {200: 1500, 404: 23, ...}
20 ip_counts = Counter() # {'[Link]': 450, ...}
21 errors = []
22
23 with open(filepath, "r", encoding="utf-8", errors="replace") as f:
24 for line_num, line in enumerate(f, 1):
25 # [Link]() tries to match at the START of the string
26 match = LOG_PATTERN.match([Link]())
27 if match:
28 data = [Link]() # Returns dict of named groups
29 status_counts[data["status"]] += 1
30 ip_counts[data["ip"]] += 1
31 else:
32 [Link](f"Line {line_num}: {line[:60]}")
33
34 return status_counts, ip_counts, errors
35
36 def report(status_counts, ip_counts, errors):
37 """Print analysis report."""
38 total = sum(status_counts.values())
39 print(f"\nTotal Requests: {total:,}")
40
41 print("\nStatus Code Breakdown:")

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 12

42 for code in sorted(status_counts):

43 count = status_counts[code]
44 bar = "#" * (count * 30 // max(status_counts.values()))
45 pct = count / total * 100
46 print(f" {code}: {count:>6,} ({pct:.1f}%) {bar}")
47
48 print("\nTop 10 Client IPs:")
49 for ip, count in ip_counts.most_common(10):
50 print(f" {ip:<20} {count:>6,} requests")
51
52 if errors:
53 print(f"\nUnparsed Lines: {len(errors)} (showing first 5)")
54 for e in errors[:5]:
55 print(f" {e}")
56
57 if __name__ == "__main__":
58 if len([Link]) < 2:
59 print("Usage: python [Link] <logfile>")
60 [Link](1)
61 sc, ic, er = parse_log([Link][1])
62 report(sc, ic, er)

■ Line-by-Line Explanation

1 [Link]()
Pre-compiles the regex pattern once. Much faster than [Link](pattern, line) in a loop which recompiles every
iteration.

2 Named groups (?P...)

Access match results by name with [Link]() instead of fragile positional indices.

3 Counter()
A dict subclass optimised for counting. counter[key] returns 0 for missing keys (no KeyError). most_common(n)
returns top-n items sorted by frequency.

4 errors='replace' in open()
Replaces undecodable bytes with the Unicode replacement character instead of crashing. Essential for real-world
log files.

5 enumerate(f, 1)
Iterates with a counter starting at 1, giving you the actual line number for error reporting.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 13

DAY 005 BEGINNER

System Health Monitor

Monitor CPU, RAM, and disk usage with threshold alerts.

■ Source Code
1 import os
2 import sys
3 import time
4
5 def get_cpu_usage():
6 """Read CPU usage from /proc/stat (Linux only)."""
7 try:
8 with open("/proc/stat") as f:
9 # First line: cpu user nice system idle iowait irq softirq
10 fields = [Link]().split()
11 # Total time = sum of all CPU time fields
12 total = sum(int(x) for x in fields[1:])
13 # Idle time is the 4th field (index 4)
14 idle = int(fields[4])
15 # Usage % = (total - idle) / total * 100
16 return (total - idle) / total * 100
17 except FileNotFoundError:
18 # /proc/stat doesn't exist on macOS/Windows
19 return -1.0
20
21 def get_memory_usage():
22 """Parse /proc/meminfo to get memory statistics."""
23 mem = {}
24 try:
25 with open("/proc/meminfo") as f:
26 for line in f:
27 # Each line: "MemTotal: 16384000 kB"
28 key, value = [Link](":", 1)
29 # Convert kB value to bytes
30 mem[[Link]()] = int([Link]().split()[0]) * 1024
31 total = [Link]("MemTotal", 0)
32 available = [Link]("MemAvailable", 0)
33 used = total - available
34 pct = used / total * 100 if total > 0 else 0
35 return total, used, pct
36 except FileNotFoundError:
37 return 0, 0, 0.0
38
39 def get_disk_usage(path="/"):
40 """Use [Link] to get disk statistics."""
41 # statvfs returns filesystem statistics for the given path

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 14

42 st = [Link](path)
43 # f_blocks * f_frsize = total size in bytes
44 total = st.f_blocks * st.f_frsize
45 # f_bavail = blocks available to non-root users
46 free = st.f_bavail * st.f_frsize
47 used = total - free
48 pct = used / total * 100 if total > 0 else 0
49 return total, used, pct
50
51 def human_readable(b):
52 for unit in ["B","KB","MB","GB","TB"]:
53 if b < 1024: return f"{b:.1f}{unit}"
54 b /= 1024
55 return f"{b:.1f}PB"
56
57 def check_thresholds(cpu_pct, mem_pct, disk_pct,
58 cpu_warn=80, mem_warn=85, disk_warn=90):
59 """Emit WARNING or CRITICAL alerts based on thresholds."""
60 alerts = []
61 if cpu_pct > cpu_warn:
62 [Link](f" [WARN] CPU usage {cpu_pct:.1f}% exceeds {cpu_warn}%")
63 if mem_pct > mem_warn:
64 [Link](f" [WARN] Memory usage {mem_pct:.1f}% exceeds {mem_warn}%")
65 if disk_pct > disk_warn:
66 [Link](f" [CRIT] Disk usage {disk_pct:.1f}% exceeds {disk_warn}%")
67 return alerts
68
69 def monitor(interval=5, iterations=3):
70 for i in range(iterations):
71 print(f"\n=== Health Check #{i+1} @ {[Link]('%H:%M:%S')} ===")
72 cpu = get_cpu_usage()
73 mt, mu, mp = get_memory_usage()
74 dt, du, dp = get_disk_usage("/")
75
76 print(f" CPU : {cpu:5.1f}%")
77 print(f" Memory: {mp:5.1f}% ({human_readable(mu)} / {human_readable(mt)})")
78 print(f" Disk : {dp:5.1f}% ({human_readable(du)} / {human_readable(dt)})")
79
80 alerts = check_thresholds(cpu, mp, dp)
81 if alerts:
82 print(" ALERTS:")
83 for a in alerts: print(a)
84 else:
85 print(" [OK] All metrics within thresholds.")
86
87 if i < iterations - 1:
88 [Link](interval)
89
90 if __name__ == "__main__":
91 monitor()

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 15

■ Line-by-Line Explanation

1 /proc/stat
Linux virtual filesystem that exposes kernel data. Reading it is zero-cost; the kernel generates the content
on-the-fly.

2 [Link]()
POSIX system call for filesystem statistics. f_bavail (not f_bfree) gives the blocks available to regular users (root
has a reserved margin).

3 Threshold-based alerting
The pattern of parameterised thresholds with warn/crit levels mirrors Nagios, Prometheus Alertmanager, and
CloudWatch Alarms.

4 [Link](interval)
Pauses the process for the specified number of seconds. Essential for polling-based monitors to avoid 100% CPU.

5 [Link]('%H:%M:%S')
Formats current local time as HH:MM:SS. Use [Link]() or [Link] for UTC timestamps in
production.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 16

PART II — FILES, DATA &

AUTOMATION (Days 31–80)

Each project in this part is a complete, runnable Python script. Read the description,

3 study the code, then check the line-by-line explanation below it. Type the code
yourself for best retention.
Projects

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 17

INTERMEDIA
DAY 031
TE

YAML Config Reader & Validator

Read, validate, and merge YAML configuration files for deployments.

■ Source Code
1 import yaml # pip install pyyaml
2 import sys
3 import os
4 from typing import Any
5
6 # Define expected schema as a dict of {key: expected_type}
7 SCHEMA = {
8 "app_name": str,
9 "version": str,
10 "replicas": int,
11 "port": int,
12 "image": str,
13 "env": dict,
14 }
15
16 def load_yaml(filepath: str) -> dict:
17 """Load and parse a YAML file safely."""
18 if not [Link](filepath):
19 raise FileNotFoundError(f"Config file not found: {filepath}")
20
21 with open(filepath, "r") as f:
22 # yaml.safe_load() prevents arbitrary Python object instantiation
23 # NEVER use [Link]() without Loader= - it is a security risk
24 data = yaml.safe_load(f)
25
26 # safe_load returns None for empty files
27 if data is None:
28 return {}
29 return data
30
31 def validate(config: dict, schema: dict) -> list:
32 """Validate config against schema. Returns list of error strings."""
33 errors = []
34 for key, expected_type in [Link]():
35 if key not in config:
36 [Link](f"Missing required key: '{key}'")
37 elif not isinstance(config[key], expected_type):
38 actual = type(config[key]).__name__
39 [Link](
40 f"'{key}': expected {expected_type.__name__}, got {actual}"

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 18

41 )
42 return errors
43
44 def merge_configs(base: dict, override: dict) -> dict:
45 """Deep-merge two configs. Override values win on conflict."""
46 result = [Link]()
47 for key, value in [Link]():
48 # If both are dicts, recurse for deep merge
49 if key in result and isinstance(result[key], dict) and isinstance(value, dict):
50 result[key] = merge_configs(result[key], value)
51 else:
52 # Otherwise override value replaces base value
53 result[key] = value
54 return result
55
56 def main():
57 base_cfg = load_yaml("[Link]")
58 env = [Link]("APP_ENV", "dev")
59 env_cfg = load_yaml(f"config.{env}.yaml") if [Link](f"config.{env}.yaml") else {}
60
61 # Merge: env-specific values override base values
62 final_cfg = merge_configs(base_cfg, env_cfg)
63
64 print(f"Loaded config for environment: {env}")
65 errors = validate(final_cfg, SCHEMA)
66
67 if errors:
68 print("\n[VALIDATION FAILED]")
69 for e in errors: print(f" - {e}")
70 [Link](1)
71
72 print("[VALID] Configuration passed all checks.")
73 print("\nEffective Configuration:")
74 # [Link]() serialises a Python dict back to YAML format
75 print([Link](final_cfg, default_flow_style=False))
76
77 if __name__ == "__main__":
78 main()

■ Line-by-Line Explanation

1 yaml.safe_load()
The ONLY safe way to load YAML. [Link]() without a Loader can execute arbitrary Python code embedded in
the YAML file — a critical security vulnerability.

2 Deep merge pattern

Shallow copy ([Link]()) only copies top-level keys. Recursive merge handles nested dicts like env: {DB_HOST:
x} correctly.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 19

3 isinstance() type checking

Validates that values match expected Python types. Critical when YAML config drives infrastructure (wrong type →
broken deployment).

4 Environment layering
base → dev/staging/prod is the 12-factor app config pattern. It avoids duplicating common settings while allowing
per-env overrides.

5 [Link](default_flow_style=False)
Produces multi-line human-readable YAML instead of the compact {key: val} inline style.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 20

DAY 032 BEGINNER

JSON to CSV Converter

Flatten nested JSON API responses into CSV for reporting and analysis.

■ Source Code
1 import json
2 import csv
3 import sys
4 import os
5 from typing import Any
6
7 def flatten_dict(d: dict, parent_key: str = "", sep: str = ".") -> dict:
8 """Flatten nested dicts: {a: {b: 1}} -> {'a.b': 1}"""
9 items = []
10 for k, v in [Link]():
11 # Build the dotted key path
12 new_key = f"{parent_key}{sep}{k}" if parent_key else k
13 if isinstance(v, dict):
14 # Recurse into nested dicts
15 [Link](flatten_dict(v, new_key, sep).items())
16 elif isinstance(v, list):
17 # Convert lists to a comma-separated string
18 # (full list flattening would explode row count)
19 [Link]((new_key, [Link](v)))
20 else:
21 [Link]((new_key, v))
22 return dict(items)
23
24 def json_to_csv(json_filepath: str, csv_filepath: str):
25 """Convert a JSON file (list of objects) to CSV."""
26 with open(json_filepath, "r") as f:
27 data = [Link](f) # [Link]() reads from file object
28
29 # Handle both a list of records and a single object
30 if isinstance(data, dict):
31 data = [data]
32
33 if not data:
34 print("Empty JSON — no CSV written.")
35 return
36
37 # Flatten each record
38 flat_records = [flatten_dict(record) for record in data]
39
40 # Collect ALL keys across all records to build a complete header
41 # Some records may have keys others don't (sparse JSON)

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 21

42 all_keys = []
43 seen = set()
44 for record in flat_records:
45 for key in [Link]():
46 if key not in seen:
47 all_keys.append(key)
48 [Link](key)
49
50 with open(csv_filepath, "w", newline="", encoding="utf-8") as f:
51 # DictWriter writes dicts as CSV rows using field names as headers
52 writer = [Link](f, fieldnames=all_keys,
53 extrasaction="ignore",
54 restval="") # restval fills missing keys
55 [Link]()
56 [Link](flat_records)
57
58 print(f"Converted {len(flat_records)} records -> {csv_filepath}")
59 print(f"Columns: {len(all_keys)}")
60
61 if __name__ == "__main__":
62 src = [Link][1] if len([Link]) > 1 else "[Link]"
63 dst = [Link][2] if len([Link]) > 2 else "[Link]"
64 json_to_csv(src, dst)

■ Line-by-Line Explanation

1 Recursive flatten
Nested JSON is the norm with API responses. Flattening with dotted paths ([Link]) preserves meaning
while making data tabular.

2 [Link](v) for lists

Storing arrays as JSON strings in a CSV cell is a pragmatic trade-off. Full explosion (one row per array element)
would require a different data model.

3 all_keys with ordering

Using a list + set combo preserves the first-seen key order (important for readability) while achieving O(1)
deduplication.

4 restval=''
When a record doesn't have a particular key, DictWriter fills the cell with this value. Without it, DictWriter raises a
ValueError for sparse records.

5 newline='' in open()
Required by the csv module documentation to prevent extra blank lines on Windows (the csv module handles its
own line termination).

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 22

INTERMEDIA
DAY 033
TE

Automated Backup Script

Create timestamped, compressed backups with retention policy.

■ Source Code
1 import os
2 import sys
3 import shutil
4 import tarfile
5 import time
6 import glob
7 from pathlib import Path
8
9 class BackupManager:
10 def __init__(self, source_dir: str, backup_dir: str, retain_count: int = 7):
11 [Link] = Path(source_dir) # [Link] for cleaner path ops
12 self.backup_dir = Path(backup_dir)
13 self.retain_count = retain_count # How many backups to keep
14
15 # Create backup directory if it doesn't exist
16 # parents=True creates intermediate directories; exist_ok=True is idempotent
17 self.backup_dir.mkdir(parents=True, exist_ok=True)
18
19 def create_backup(self) -> Path:
20 """Create a gzip-compressed tar archive of the source directory."""
21 # Timestamp format: YYYYMMDD_HHMMSS — lexicographic = chronological order
22 timestamp = [Link]("%Y%m%d_%H%M%S")
23 archive_name = f"backup_{[Link]}_{timestamp}.[Link]"
24 archive_path = self.backup_dir / archive_name # Path / operator joins paths
25
26 print(f"Creating backup: {archive_name}")
27
28 # [Link] with 'w:gz' mode creates a gzip-compressed archive
29 # 'w:bz2' for bzip2 (smaller but slower), 'w:xz' for xz (smallest)
30 with [Link](archive_path, "w:gz") as tar:
31 # arcname sets the root directory name inside the archive
32 [Link](str([Link]), arcname=[Link])
33
34 size = archive_path.stat().st_size
35 print(f" Backup created: {size / 1024 / 1024:.1f} MB")
36 return archive_path
37
38 def apply_retention(self):
39 """Delete old backups, keeping only the N most recent."""
40 # glob to find all backup archives, sorted by name (= by time)

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 23

41 pattern = str(self.backup_dir / f"backup_{[Link]}_*.[Link]")

42 backups = sorted([Link](pattern))
43
44 to_delete = backups[:-self.retain_count] if len(backups) > self.retain_count else []
45
46 for old_backup in to_delete:
47 [Link](old_backup)
48 print(f" Deleted old backup: {Path(old_backup).name}")
49
50 remaining = len(backups) - len(to_delete)
51 print(f" Retention: {remaining}/{self.retain_count} backups kept")
52
53 def verify_backup(self, archive_path: Path) -> bool:
54 """Verify archive integrity by testing it without extracting."""
55 try:
56 with [Link](archive_path, "r:gz") as tar:
57 # getmembers() reads the entire archive index
58 members = [Link]()
59 print(f" Verified: {len(members)} files in archive")
60 return True
61 except [Link] as e:
62 print(f" CORRUPT BACKUP: {e}")
63 return False
64
65 def run(self):
66 """Full backup cycle: create, verify, apply retention."""
67 archive = self.create_backup()
68 ok = self.verify_backup(archive)
69 if not ok:
70 [Link](1)
71 self.apply_retention()
72 print(" Backup cycle complete.")
73
74 if __name__ == "__main__":
75 src = [Link][1] if len([Link]) > 1 else "/etc"
76 dst = [Link][2] if len([Link]) > 2 else "/tmp/backups"
77 manager = BackupManager(src, dst, retain_count=7)
78 [Link]()

■ Line-by-Line Explanation

1 [Link]
The modern way to handle paths in Python 3. Path('/tmp') / 'backups' / '[Link]' is safer and clearer than
[Link]().

2 [Link]('w:gz')
Creates a new gzip-compressed archive. The 'w:gz' mode string is more explicit than relying on filename
extension.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 24

3 Lexicographic timestamp YYYYMMDD_HHMMSS

When filenames sort alphabetically, they also sort chronologically. This is why ISO 8601 date format is standard in
DevOps tooling.

4 Retention by slice backups[:-N]

Negative indexing: backups[:-7] returns everything except the last 7 items. An elegant one-liner for retention policy.

5 Verify before apply retention

Always verify the new backup BEFORE deleting old ones. If the backup is corrupt, you still have previous good
copies.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 25

PART III — NETWORKING & APIs

(Days 81–130)

Each project in this part is a complete, runnable Python script. Read the description,

2 study the code, then check the line-by-line explanation below it. Type the code
yourself for best retention.
Projects

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 26

INTERMEDIA
DAY 081
TE

HTTP Health Checker

Ping a list of HTTP endpoints and report their status, latency, and SSL expiry.

■ Source Code
1 import [Link] # Built-in HTTP client (no extra install needed)
2 import [Link]
3 import ssl
4 import socket
5 import time
6 import json
7 from datetime import datetime
8
9 ENDPOINTS = [
10 "[Link]
11 "[Link]
12 "[Link]
13 ]
14
15 def check_ssl_expiry(hostname: str, port: int = 443) -> int:
16 """Return days until SSL certificate expires. -1 if error."""
17 ctx = ssl.create_default_context()
18 try:
19 with ctx.wrap_socket([Link](), server_hostname=hostname) as s:
20 # Set a 5 second timeout to avoid hanging
21 [Link](5)
22 [Link]((hostname, port))
23 cert = [Link]() # Returns dict of certificate fields
24 # notAfter format: "Oct 1 00:00:00 2025 GMT"
25 expiry = [Link](cert["notAfter"], "%b %d %H:%M:%S %Y %Z")
26 return (expiry - [Link]()).days
27 except Exception:
28 return -1
29
30 def check_endpoint(url: str, timeout: int = 10) -> dict:
31 """Check a single endpoint and return a result dict."""
32 result = {
33 "url": url,
34 "status": None,
35 "latency_ms": None,
36 "ssl_days": None,
37 "error": None,
38 }
39
40 start = [Link]() # monotonic clock is immune to system time changes

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 27

41
42 try:
43 # [Link] opens the URL and returns a response object
44 req = [Link](url, headers={"User-Agent": "DevOps-Checker/1.0"})
45 with [Link](req, timeout=timeout) as resp:
46 result["status"] = [Link] # HTTP status code (200, 404, etc.)
47 result["latency_ms"] = round(([Link]() - start) * 1000, 1)
48
49 except [Link] as e:
50 # HTTPError is raised for 4xx/5xx responses (still a valid response)
51 result["status"] = [Link]
52 result["latency_ms"] = round(([Link]() - start) * 1000, 1)
53
54 except [Link] as e:
55 # URLError covers DNS failures, connection refused, timeouts
56 result["error"] = str([Link])
57
58 # Check SSL for https URLs
59 if [Link]("[Link]
60 # Extract hostname from URL (crude but effective)
61 hostname = [Link]("//")[1].split("/")[0]
62 result["ssl_days"] = check_ssl_expiry(hostname)
63
64 return result
65
66 def report(results: list):
67 print(f"\n{'URL':<40} {'Status':>7} {'Latency':>10} {'SSL Days':>10} {'Note'}")
68 print("-" * 80)
69 for r in results:
70 ssl_str = str(r["ssl_days"]) if r["ssl_days"] is not None else "N/A"
71 note = r["error"] or ("WARNING: cert expires soon" if r["ssl_days"] and r["ssl_days"] < 30 else
"OK")
72 lat = f"{r['latency_ms']}ms" if r["latency_ms"] else "N/A"
73 print(f"{r['url']:<40} {str(r['status']):>7} {lat:>10} {ssl_str:>10} {note}")
74
75 if __name__ == "__main__":
76 results = [check_endpoint(url) for url in ENDPOINTS]
77 report(results)
78 # Exit non-zero if any endpoint is not 200
79 if any(r["status"] != 200 for r in results):
80 import sys; [Link](1)

■ Line-by-Line Explanation

1 [Link]()
Unlike [Link](), the monotonic clock never goes backwards (immune to NTP adjustments). Always use it for
measuring durations.

2 [Link]
Allows setting custom headers. User-Agent is important: some servers block requests without it.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 28

3 [Link] vs URLError
HTTPError is a subclass of URLError. Catch HTTPError first to handle 4xx/5xx responses (you still get the status
code). URLError catches network-level failures.

4 ssl.create_default_context() + wrap_socket()
The standard way to establish a TLS connection and inspect the certificate without making a full HTTP request.

5 [Link](1) at the end

Makes the script usable as a CI/CD gate: any non-200 response fails the pipeline step.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 29

INTERMEDIA
DAY 082
TE

REST API Client with Retry

A robust API client with exponential backoff, rate-limit handling, and response caching.

■ Source Code
1 import [Link]
2 import [Link]
3 import [Link]
4 import json
5 import time
6 import os
7 import hashlib
8
9 class APIClient:
10 """Reusable REST API client with retry, backoff, and caching."""
11
12 def __init__(self, base_url: str, api_key: str = None,
13 max_retries: int = 3, cache_dir: str = None):
14 self.base_url = base_url.rstrip("/")
15 self.api_key = api_key
16 self.max_retries = max_retries
17 self.cache_dir = cache_dir
18
19 if cache_dir:
20 [Link](cache_dir, exist_ok=True)
21
22 def _cache_key(self, url: str, params: dict) -> str:
23 """Generate a unique cache key from URL + params."""
24 cache_str = url + [Link](params or {}, sort_keys=True)
25 # MD5 is fast; we're not using it for security here
26 return hashlib.md5(cache_str.encode()).hexdigest()
27
28 def _get_cached(self, key: str, ttl_seconds: int = 300):
29 """Return cached data if fresh, else None."""
30 if not self.cache_dir: return None
31 path = [Link](self.cache_dir, f"{key}.json")
32 if not [Link](path): return None
33 age = [Link]() - [Link](path)
34 if age > ttl_seconds: return None
35 with open(path) as f:
36 return [Link](f)
37
38 def _set_cached(self, key: str, data):
39 """Write data to the cache."""
40 if not self.cache_dir: return

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 30

41 path = [Link](self.cache_dir, f"{key}.json")

42 with open(path, "w") as f:
43 [Link](data, f)
44
45 def get(self, endpoint: str, params: dict = None,
46 ttl: int = 300) -> dict:
47 """Make a GET request with retry and exponential backoff."""
48 url = f"{self.base_url}/{[Link]('/')}"
49 if params:
50 url += "?" + [Link](params)
51
52 # Check cache before making network request
53 ck = self._cache_key(url, params)
54 cached = self._get_cached(ck, ttl)
55 if cached is not None:
56 print(f" [CACHE HIT] {endpoint}")
57 return cached
58
59 for attempt in range(1, self.max_retries + 1):
60 try:
61 headers = {"Accept": "application/json"}
62 if self.api_key:
63 headers["Authorization"] = f"Bearer {self.api_key}"
64
65 req = [Link](url, headers=headers)
66 with [Link](req, timeout=15) as resp:
67 data = [Link]([Link]().decode("utf-8"))
68 self._set_cached(ck, data)
69 return data
70
71 except [Link] as e:
72 if [Link] == 429:
73 # 429 Too Many Requests — respect the Retry-After header
74 retry_after = int([Link]("Retry-After", 60))
75 print(f" Rate limited. Waiting {retry_after}s...")
76 [Link](retry_after)
77 elif [Link] >= 500 and attempt < self.max_retries:
78 # 5xx = server error, worth retrying
79 wait = 2 ** attempt # Exponential backoff: 2, 4, 8...
80 print(f" Attempt {attempt} failed (HTTP {[Link]}). Retry in {wait}s")
81 [Link](wait)
82 else:
83 raise # 4xx client errors: don't retry
84
85 except [Link] as e:
86 if attempt < self.max_retries:
87 wait = 2 ** attempt
88 print(f" Network error: {[Link]}. Retry in {wait}s")
89 [Link](wait)
90 else:

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 31

91 raise
92
93 raise RuntimeError(f"All {self.max_retries} attempts failed for {endpoint}")
94
95 if __name__ == "__main__":
96 client = APIClient("[Link] cache_dir="/tmp/api_cache")
97 # Fetch GitHub user info
98 user = [Link]("/users/torvalds")
99 print(f"Name : {[Link]('name')}")
100 print(f"Public Repos: {[Link]('public_repos')}")
101 print(f"Followers : {[Link]('followers')}")

■ Line-by-Line Explanation

1 Exponential backoff 2**attempt

After failed attempt 1: wait 2s, attempt 2: wait 4s, attempt 3: wait 8s. This prevents thundering-herd problems when
multiple clients retry simultaneously.

2 HTTP 429 handling

Respecting Retry-After is mandatory for well-behaved API clients. Ignoring it leads to IP bans.

3 File-based TTL cache

Checking [Link]() vs [Link]() is a simple but effective cache expiry mechanism. No Redis needed for
scripts.

4 hashlib.md5 for cache keys

Not for security, just for generating short, consistent filenames from arbitrary URL strings.

5 raise vs raise RuntimeError

Re-raising the original exception preserves the full stack trace. Only raise a new exception when you need to add
context.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 32

PART IV — DOCKER & CONTAINERS

(Days 131–180)

Each project in this part is a complete, runnable Python script. Read the description,

2 study the code, then check the line-by-line explanation below it. Type the code
yourself for best retention.
Projects

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 33

INTERMEDIA
DAY 131
TE

Docker Container Inspector

Query the Docker daemon via its Unix socket API to list and inspect containers.

■ Source Code
1 import socket
2 import json
3 import os
4 import sys
5
6 class DockerClient:
7 """Minimal Docker API client using the Unix socket directly."""
8
9 SOCKET_PATH = "/var/run/[Link]" # Default Docker daemon socket
10
11 def __init__(self):
12 if not [Link](self.SOCKET_PATH):
13 raise RuntimeError(
14 f"Docker socket not found at {self.SOCKET_PATH}\n"
15 "Is Docker running? Are you in the docker group?"
16 )
17
18 def _request(self, method: str, path: str) -> dict:
19 """Send an HTTP request over the Unix domain socket."""
20 # Unix domain socket is like TCP but uses a file instead of IP:port
21 sock = [Link](socket.AF_UNIX, socket.SOCK_STREAM)
22 [Link](10)
23
24 try:
25 [Link](self.SOCKET_PATH)
26
27 # HTTP/1.0 request format: METHOD /path HTTP/1.0\r\nHost: localhost\r\n\r\n
28 request = (
29 f"{method} {path} HTTP/1.0\r\n"
30 f"Host: localhost\r\n"
31 f"Accept: application/json\r\n"
32 f"\r\n"
33 )
34 [Link]([Link]())
35
36 # Read the full response into memory
37 response = b""
38 while True:
39 chunk = [Link](4096)
40 if not chunk:

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 34

41 break
42 response += chunk
43 finally:
44 [Link]() # Always close the socket
45
46 # Split HTTP headers from body at \r\n\r\n boundary
47 header_end = [Link](b"\r\n\r\n")
48 body = response[header_end + 4:]
49
50 return [Link]([Link]("utf-8"))
51
52 def list_containers(self, all_containers: bool = False) -> list:
53 """List containers (running only by default)."""
54 params = "?all=true" if all_containers else ""
55 return self._request("GET", f"/containers/json{params}")
56
57 def inspect_container(self, container_id: str) -> dict:
58 """Get full details of a specific container."""
59 return self._request("GET", f"/containers/{container_id}/json")
60
61 def get_stats(self, container_id: str) -> dict:
62 """Get real-time resource stats for a container (single snapshot)."""
63 return self._request("GET", f"/containers/{container_id}/stats?stream=false")
64
65 def format_bytes(b):
66 for unit in ["B","KB","MB","GB"]:
67 if b < 1024: return f"{b:.0f}{unit}"
68 b /= 1024
69 return f"{b:.0f}TB"
70
71 def main():
72 client = DockerClient()
73 containers = client.list_containers(all_containers=True)
74
75 if not containers:
76 print("No containers found.")
77 return
78
79 print(f"{'ID':<14} {'Name':<30} {'Image':<25} {'Status':<15} {'CPU%':>7} {'Memory':>10}")
80 print("-" * 105)
81
82 for c in containers:
83 cid = c["Id"][:12]
84 name = c["Names"][0].lstrip("/") if c["Names"] else "unknown"
85 image = c["Image"][:24]
86 status = c["Status"][:14]
87
88 # Only get stats for running containers
89 cpu_pct = "-"
90 mem_str = "-"

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 35

91 if c["State"] == "running":
92 try:
93 stats = client.get_stats(c["Id"])
94 # CPU % calculation from Docker stats API
95 cpu_delta = stats["cpu_stats"]["cpu_usage"]["total_usage"] - \
96 stats["precpu_stats"]["cpu_usage"]["total_usage"]
97 sys_delta = stats["cpu_stats"]["system_cpu_usage"] - \
98 stats["precpu_stats"]["system_cpu_usage"]
99 num_cpus = stats["cpu_stats"].get("online_cpus", 1)
100 if sys_delta > 0:
101 cpu_pct = f"{(cpu_delta / sys_delta) * num_cpus * 100:.1f}%"
102 mem_used = stats["memory_stats"].get("usage", 0)
103 mem_str = format_bytes(mem_used)
104 except Exception:
105 pass
106
107 print(f"{cid:<14} {name:<30} {image:<25} {status:<15} {cpu_pct:>7} {mem_str:>10}")
108
109 if __name__ == "__main__":
110 main()

■ Line-by-Line Explanation

1 Unix domain socket (AF_UNIX)

Docker's API is an HTTP API served over /var/run/[Link] instead of a TCP port. AF_UNIX sockets
communicate via the filesystem, not the network.

2 HTTP/1.0 over raw socket

Docker accepts standard HTTP. We manually construct the request string rather than using a library to
demonstrate what HTTP actually is.

3 CPU % calculation
Docker provides raw nanosecond CPU counters. The delta between two samples divided by the system delta gives
the CPU percentage. This mirrors how docker stats computes it.

4 stream=false for stats

By default, /stats streams continuously. Adding ?stream=false returns a single snapshot, which is what we need for
a point-in-time check.

5 [Link](10)
Prevents the script from hanging indefinitely if the Docker daemon is slow or unresponsive.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 36

INTERMEDIA
DAY 132
TE

Dockerfile Linter

Statically analyze a Dockerfile for common security and best-practice violations.

■ Source Code
1 import sys
2 import re
3 from dataclasses import dataclass, field
4 from typing import List, Optional
5
6 @dataclass
7 class LintResult:
8 line_num: int
9 severity: str # ERROR, WARN, INFO
10 rule_id: str
11 message: str
12 line_text: str
13
14 class DockerfileLinter:
15 """Static analyzer for Dockerfiles."""
16
17 # Rules: list of (regex_pattern, severity, rule_id, message_template)
18 RULES = [
19 # Security rules
20 (r"FROM\s+.*:latest", "WARN", "DL3007",
21 "Using 'latest' tag. Pin image to a specific version for reproducibility."),
22 (r"FROM\s+ubuntu(?::.*)?$|FROM\s+debian(?::.*)?$", "INFO", "DL3006",
23 "Consider using a minimal base image (alpine, distroless) to reduce attack surface."),
24 (r"RUN\s+.*sudo\s", "WARN", "DL3004",
25 "Do not use sudo in RUN; the container runs as the given USER already."),
26 (r"RUN\s+.*apt-get\s+install(?!.*-y)", "ERROR", "DL3015",
27 "apt-get install without -y flag may hang waiting for user input."),
28 (r"RUN\s+.*apt-get\s+update[^&\n]*$", "WARN", "DL3009",
29 "apt-get update without apt-get install in same RUN creates stale cache."),
30 (r"ADD\s+https?://", "WARN", "DL3020",
31 "Use COPY instead of ADD for local files. Use curl/wget in RUN for URLs."),
32 (r"(PASSWORD|SECRET|API_KEY|TOKEN)\s*=\s*\S+", "ERROR", "DL4006",
33 "Potential hardcoded secret. Use --build-arg or multi-stage builds."),
34 # Best practice rules
35 (r"^RUN\s", "INFO", "DL3059",
36 "Multiple consecutive RUN instructions. Combine to reduce layers."),
37 (r"MAINTAINER\s", "WARN", "DL4000",
38 "MAINTAINER is deprecated. Use LABEL maintainer= instead."),
39 ]
40

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 37

41 def lint(self, dockerfile_path: str) -> List[LintResult]:

42 results = []
43
44 with open(dockerfile_path, "r") as f:
45 lines = [Link]()
46
47 prev_was_run = False
48 has_user_instruction = False
49 has_healthcheck = False
50
51 for line_num, line in enumerate(lines, 1):
52 stripped = [Link]()
53
54 # Skip empty lines and comments
55 if not stripped or [Link]("#"):
56 continue
57
58 # Track USER instruction
59 if [Link]().startswith("USER "):
60 has_user_instruction = True
61 if [Link]().startswith("HEALTHCHECK "):
62 has_healthcheck = True
63
64 # Check consecutive RUN instructions
65 is_run = [Link]().startswith("RUN ")
66 if is_run and prev_was_run:
67 [Link](LintResult(
68 line_num=line_num, severity="INFO",
69 rule_id="DL3059",
70 message="Consider combining with previous RUN to reduce layers.",
71 line_text=stripped[:60]
72 ))
73 prev_was_run = is_run
74
75 # Apply regex rules
76 for pattern, severity, rule_id, message in [Link]:
77 if rule_id == "DL3059": continue # Already handled above
78 if [Link](pattern, stripped, [Link]):
79 [Link](LintResult(
80 line_num=line_num, severity=severity,
81 rule_id=rule_id, message=message,
82 line_text=stripped[:60]
83 ))
84
85 # File-level checks
86 if not has_user_instruction:
87 [Link](LintResult(
88 line_num=0, severity="WARN", rule_id="DL3002",
89 message="No USER instruction. Container will run as root.",
90 line_text=""

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 38

91 ))
92 if not has_healthcheck:
93 [Link](LintResult(
94 line_num=0, severity="INFO", rule_id="DL3029",
95 message="No HEALTHCHECK instruction defined.",
96 line_text=""
97 ))
98
99 return sorted(results, key=lambda r: r.line_num)
100
101 def report(self, results: List[LintResult]) -> bool:
102 """Print results and return True if no errors."""
103 icons = {"ERROR": "[E]", "WARN": "[W]", "INFO": "[I]"}
104 errors = 0
105 for r in results:
106 loc = f"Line {r.line_num:>3}" if r.line_num > 0 else " (file)"
107 print(f"{icons[[Link]]} {loc} [{r.rule_id}] {[Link]}")
108 if r.line_text:
109 print(f" > {r.line_text}")
110 if [Link] == "ERROR":
111 errors += 1
112
113 print(f"\nTotal: {len(results)} findings ({errors} errors)")
114 return errors == 0
115
116 if __name__ == "__main__":
117 path = [Link][1] if len([Link]) > 1 else "Dockerfile"
118 linter = DockerfileLinter()
119 results = [Link](path)
120 ok = [Link](results)
121 [Link](0 if ok else 1)

■ Line-by-Line Explanation

1 @dataclass
Automatically generates __init__, __repr__, and __eq__ methods. Cleaner than writing boilerplate classes for data
containers.

2 [Link] vs [Link]
search() finds the pattern anywhere in the string; match() only looks at the start. Use search() for line scanning.

3 Sorted results by line_num

Output ordered by line number mimics how editors display lint warnings, making them actionable.

4 File-level checks after the loop

Some rules (no USER, no HEALTHCHECK) apply to the entire file, not a specific line. Run these after processing
all lines.

5 [Link](0 if ok else 1)
Zero means success, non-zero means failure. This makes the linter usable in CI/CD pipelines as a quality gate.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 39

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 40

PART V — AWS & CLOUD (Days

181–230)

Each project in this part is a complete, runnable Python script. Read the description,

2 study the code, then check the line-by-line explanation below it. Type the code
yourself for best retention.
Projects

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 41

INTERMEDIA
DAY 181
TE

AWS S3 Bucket Manager

List, upload, download, and delete S3 objects using boto3.

■ Source Code
1 import boto3 # pip install boto3
2 import os
3 import sys
4 import time
5 from [Link] import ClientError, NoCredentialsError
6
7 class S3Manager:
8 """High-level S3 operations wrapper with progress reporting."""
9
10 def __init__(self, region: str = None):
11 # [Link]() reads credentials from:
12 # 1. Environment: AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY
13 # 2. ~/.aws/credentials file
14 # 3. IAM role (when running on EC2/ECS/Lambda)
15 try:
16 self.s3 = [Link]("s3", region_name=region)
17 # Test credentials by making a cheap API call
18 self.s3.list_buckets()
19 except NoCredentialsError:
20 print("ERROR: No AWS credentials found.")
21 print("Set AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY, or run aws configure")
22 [Link](1)
23
24 def list_buckets(self) -> list:
25 """Return list of all S3 buckets in the account."""
26 response = self.s3.list_buckets()
27 # Response is a dict; 'Buckets' key contains the list
28 return [Link]("Buckets", [])
29
30 def list_objects(self, bucket: str, prefix: str = "") -> list:
31 """List objects in a bucket, handling pagination automatically."""
32 objects = []
33 paginator = self.s3.get_paginator("list_objects_v2")
34 # Paginators automatically handle NextContinuationToken
35 # without them you'd need a while loop checking IsTruncated
36 for page in [Link](Bucket=bucket, Prefix=prefix):
37 [Link]([Link]("Contents", []))
38 return objects
39
40 def upload_file(self, local_path: str, bucket: str, s3_key: str = None):

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 42

41 """Upload a file with progress callback."""

42 if s3_key is None:
43 s3_key = [Link](local_path)
44
45 file_size = [Link](local_path)
46 uploaded = [0] # Mutable container to allow closure modification
47
48 def progress_callback(bytes_transferred):
49 uploaded[0] += bytes_transferred
50 pct = uploaded[0] / file_size * 100
51 # \r overwrites the current line (no newline)
52 print(f"\r Uploading: {pct:.1f}% ({uploaded[0]:,}/{file_size:,} bytes)",
53 end="", flush=True)
54
55 try:
56 # upload_file handles multipart upload automatically for large files
57 # Callback is called for each chunk transferred
58 self.s3.upload_file(local_path, bucket, s3_key,
59 Callback=progress_callback)
60 print(f"\n Uploaded: s3://{bucket}/{s3_key}")
61 except ClientError as e:
62 # ClientError wraps all AWS API errors
63 print(f"\n Upload failed: {[Link]['Error']['Message']}")
64 raise
65
66 def download_file(self, bucket: str, s3_key: str, local_path: str):
67 """Download an S3 object to a local path."""
68 [Link]([Link](local_path) or ".", exist_ok=True)
69 self.s3.download_file(bucket, s3_key, local_path)
70 print(f" Downloaded: s3://{bucket}/{s3_key} -> {local_path}")
71
72 def delete_object(self, bucket: str, s3_key: str):
73 """Delete a single S3 object."""
74 self.s3.delete_object(Bucket=bucket, Key=s3_key)
75 print(f" Deleted: s3://{bucket}/{s3_key}")
76
77 def get_bucket_size(self, bucket: str) -> tuple:
78 """Return (total_objects, total_size_bytes) for a bucket."""
79 objects = self.list_objects(bucket)
80 total_size = sum([Link]("Size", 0) for obj in objects)
81 return len(objects), total_size
82
83 def main():
84 manager = S3Manager(region="us-east-1")
85
86 print("=== S3 Buckets ===")
87 buckets = manager.list_buckets()
88 for b in buckets:
89 print(f" {b['Name']:<40} Created: {b['CreationDate'].strftime('%Y-%m-%d')}")
90

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 43

91 if buckets:
92 bucket_name = buckets[0]["Name"]
93 count, size = manager.get_bucket_size(bucket_name)
94 print(f"\n Bucket '{bucket_name}': {count} objects, {size/1024/1024:.1f} MB")
95
96 if __name__ == "__main__":
97 main()

■ Line-by-Line Explanation

1 boto3 credential chain

boto3 checks multiple sources in order: env vars → ~/.aws/credentials → IAM role. This lets the same code work
locally and in production without code changes.

2 Paginators
AWS APIs return a maximum of 1000 items per call. Paginators handle the NextContinuationToken loop
automatically. Always use paginators for list operations.

3 Progress callback
S3 upload_file accepts a Callback parameter called for each chunk. The closure captures the uploaded list
(mutable) to track cumulative progress.

4 ClientError wrapping
All AWS API errors are ClientError exceptions. [Link]['Error']['Code'] gives the specific error code (e.g.,
'AccessDenied', 'NoSuchBucket').

5 Multipart upload
upload_file automatically uses multipart upload for files > 8MB, enabling resumable uploads and parallel part
transfers.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 44

INTERMEDIA
DAY 182
TE

EC2 Instance Manager

List, start, stop, and tag EC2 instances programmatically.

■ Source Code
1 import boto3
2 import sys
3 import time
4 from [Link] import ClientError
5
6 class EC2Manager:
7 def __init__(self, region: str = "us-east-1"):
8 # ec2 resource provides object-oriented access to instances
9 self.ec2 = [Link]("ec2", region_name=region)
10 # ec2 client provides direct API access for operations not on resource
11 [Link] = [Link]("ec2", region_name=region)
12 [Link] = region
13
14 def list_instances(self, filters: list = None) -> list:
15 """List instances with optional filters."""
16 if filters is None:
17 filters = []
18 # [Link]() translates to DescribeInstances API call
19 instances = list([Link](Filters=filters))
20 return instances
21
22 def get_instance_name(self, instance) -> str:
23 """Extract the Name tag from an instance."""
24 if [Link]:
25 for tag in [Link]:
26 if tag["Key"] == "Name":
27 return tag["Value"]
28 return "(no name)"
29
30 def print_instances(self, instances: list):
31 """Print a formatted table of instances."""
32 print(f"\n{'ID':<20} {'Name':<25} {'Type':<14} {'State':<12} {'IP':<16}")
33 print("-" * 90)
34 for inst in instances:
35 name = self.get_instance_name(inst)
36 ip = inst.public_ip_address or inst.private_ip_address or "N/A"
37 print(f"{[Link]:<20} {name:<25} {inst.instance_type:<14} "
38 f"{[Link]['name']:<12} {ip:<16}")
39
40 def start_instance(self, instance_id: str, wait: bool = True):

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 45

41 """Start a stopped instance and optionally wait for running state."""

42 instance = [Link](instance_id)
43 print(f"Starting {instance_id}...")
44
45 response = [Link]()
46 print(f" State change: {response['StartingInstances'][0]['PreviousState']['Name']} "
47 f"-> {response['StartingInstances'][0]['CurrentState']['Name']}")
48
49 if wait:
50 print(" Waiting for running state...", end="", flush=True)
51 # Waiter polls the API every 15 seconds up to 40 times (10 min max)
52 waiter = [Link].get_waiter("instance_running")
53 [Link](InstanceIds=[instance_id])
54 print(" done")
55 [Link]() # Refresh the instance object after state change
56 print(f" Public IP: {instance.public_ip_address}")
57
58 def stop_instance(self, instance_id: str, wait: bool = False):
59 """Stop a running instance."""
60 instance = [Link](instance_id)
61 [Link]()
62 print(f"Stopping {instance_id}...")
63 if wait:
64 waiter = [Link].get_waiter("instance_stopped")
65 [Link](InstanceIds=[instance_id])
66 print(f" {instance_id} is now stopped.")
67
68 def tag_instance(self, instance_id: str, tags: dict):
69 """Apply tags to an instance."""
70 # Tags are a list of {'Key': k, 'Value': v} dicts
71 tag_list = [{"Key": k, "Value": v} for k, v in [Link]()]
72 [Link].create_tags(Resources=[instance_id], Tags=tag_list)
73 print(f"Tagged {instance_id}: {tags}")
74
75 def find_untagged_instances(self) -> list:
76 """Find instances missing the Name tag — useful for compliance audits."""
77 all_instances = self.list_instances()
78 untagged = []
79 for inst in all_instances:
80 names = [t["Value"] for t in ([Link] or []) if t["Key"] == "Name"]
81 if not names:
82 [Link](inst)
83 return untagged
84
85 if __name__ == "__main__":
86 mgr = EC2Manager()
87 instances = mgr.list_instances()
88 mgr.print_instances(instances)
89
90 untagged = mgr.find_untagged_instances()

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 46

91 if untagged:
92 print(f"\nWARNING: {len(untagged)} untagged instance(s) found:")
93 for inst in untagged:
94 print(f" {[Link]} ({inst.instance_type}) - {[Link]['name']}")

■ Line-by-Line Explanation

1 [Link] vs [Link]
resource gives object-oriented access ([Link]()), while client is a thin wrapper over raw API calls. Use
resource for simplicity, client when you need full API control.

2 [Link] iteration
Tags in AWS are always a list of Key/Value dicts, never a plain dict. This is because the same key can technically
appear multiple times.

3 Waiters
Built-in polling loops that wait for a resource to reach a desired state. Use them instead of writing your own
sleep/retry loops.

4 [Link]()
After state changes, the instance object's cached attributes may be stale. reload() fetches fresh data from the API.

5 Compliance audit pattern

Searching for untagged resources is a common DevOps/FinOps task. Tags enable cost allocation, automation
targeting, and policy enforcement.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 47

PART VI — CI/CD & AUTOMATION

(Days 231–280)

Each project in this part is a complete, runnable Python script. Read the description,

1 study the code, then check the line-by-line explanation below it. Type the code
yourself for best retention.
Projects

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 48

DAY 231 ADVANCED

GitHub Actions Workflow Validator

Parse and validate GitHub Actions YAML workflow files for correctness.

■ Source Code
1 import yaml
2 import os
3 import sys
4 from pathlib import Path
5 from typing import List, Dict, Any
6
7 class WorkflowValidator:
8 """Validates GitHub Actions workflow YAML files."""
9
10 VALID_EVENTS = {
11 "push", "pull_request", "workflow_dispatch", "schedule",
12 "release", "issues", "issue_comment", "create", "delete",
13 "deployment", "fork", "gollum", "label", "milestone",
14 "page_build", "project", "public", "registry_package",
15 "repository_dispatch", "status", "watch", "workflow_call",
16 "workflow_run",
17 }
18
19 VALID_PERMISSIONS = {
20 "actions", "checks", "contents", "deployments", "id-token",
21 "issues", "discussions", "packages", "pages", "pull-requests",
22 "repository-projects", "security-events", "statuses",
23 }
24
25 def __init__(self):
26 [Link]: List[str] = []
27 [Link]: List[str] = []
28
29 def _err(self, msg: str): [Link](f"[ERROR] {msg}")
30 def _warn(self, msg: str): [Link](f"[WARN] {msg}")
31
32 def validate(self, filepath: str) -> bool:
33 """Main validation entry point. Returns True if valid."""
34 [Link] = []
35 [Link] = []
36
37 try:
38 with open(filepath) as f:
39 workflow = yaml.safe_load(f)
40 except [Link] as e:
41 self._err(f"YAML parse error: {e}")

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 49

42 return False
43
44 if not isinstance(workflow, dict):
45 self._err("Workflow must be a YAML mapping (dict).")
46 return False
47
48 self._validate_name(workflow)
49 self._validate_on(workflow)
50 self._validate_jobs(workflow)
51 self._validate_permissions(workflow)
52
53 return len([Link]) == 0
54
55 def _validate_name(self, wf: dict):
56 if "name" not in wf:
57 self._warn("No 'name' key. Add a descriptive workflow name.")
58
59 def _validate_on(self, wf: dict):
60 """Validate the 'on' trigger section."""
61 on = [Link]("on") or [Link](True) # YAML parses 'on' as True!
62 if on is None:
63 self._err("Missing 'on:' trigger. Workflow will never run.")
64 return
65
66 if isinstance(on, str):
67 on = {on: None}
68 elif isinstance(on, list):
69 on = {event: None for event in on}
70
71 if isinstance(on, dict):
72 for event in [Link]():
73 if str(event) not in self.VALID_EVENTS:
74 self._err(f"Unknown trigger event: '{event}'")
75
76 def _validate_jobs(self, wf: dict):
77 """Validate the jobs section."""
78 jobs = [Link]("jobs")
79 if not jobs:
80 self._err("No 'jobs' defined. Workflow does nothing.")
81 return
82
83 if not isinstance(jobs, dict):
84 self._err("'jobs' must be a mapping.")
85 return
86
87 all_job_ids = set([Link]())
88
89 for job_id, job in [Link]():
90 prefix = f"Job '{job_id}'"
91

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 50

92 if not isinstance(job, dict):

93 self._err(f"{prefix}: must be a mapping.")
94 continue
95
96 # Every job needs runs-on
97 if "runs-on" not in job:
98 self._err(f"{prefix}: missing 'runs-on' (e.g. ubuntu-latest).")
99
100 # Check needs references
101 needs = [Link]("needs", [])
102 if isinstance(needs, str): needs = [needs]
103 for dep in needs:
104 if dep not in all_job_ids:
105 self._err(f"{prefix}: 'needs' references unknown job '{dep}'.")
106
107 # Validate steps
108 steps = [Link]("steps", [])
109 if not steps:
110 self._warn(f"{prefix}: has no steps.")
111 else:
112 self._validate_steps(job_id, steps)
113
114 def _validate_steps(self, job_id: str, steps: list):
115 """Validate individual steps within a job."""
116 for i, step in enumerate(steps, 1):
117 prefix = f"Job '{job_id}' Step {i}"
118
119 if not isinstance(step, dict):
120 self._err(f"{prefix}: step must be a mapping.")
121 continue
122
123 has_uses = "uses" in step
124 has_run = "run" in step
125
126 # A step must have either 'uses' or 'run', not both
127 if has_uses and has_run:
128 self._err(f"{prefix}: cannot have both 'uses' and 'run'.")
129 elif not has_uses and not has_run:
130 self._err(f"{prefix}: must have either 'uses' or 'run'.")
131
132 # Warn about unpinned actions
133 if has_uses:
134 action = step["uses"]
135 if "@" not in action and not [Link]("./"):
136 self._warn(f"{prefix}: action '{action}' should be pinned to a SHA.")
137
138 def _validate_permissions(self, wf: dict):
139 """Check that permissions are minimal (security best practice)."""
140 perms = [Link]("permissions")
141 if isinstance(perms, str) and perms == "write-all":

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 51

142 self._warn("Top-level 'permissions: write-all' is too broad. Restrict to minimum.")

143
144 def report(self):
145 for w in [Link]: print(w)
146 for e in [Link]: print(e)
147 print(f"\nResult: {len([Link])} errors, {len([Link])} warnings")
148
149 if __name__ == "__main__":
150 path = [Link][1] if len([Link]) > 1 else ".github/workflows/[Link]"
151 v = WorkflowValidator()
152 ok = [Link](path)
153 [Link]()
154 [Link](0 if ok else 1)

■ Line-by-Line Explanation

1 yaml.safe_load parses 'on' as True

This is a famous YAML gotcha: 'on' without quotes is parsed as the boolean True in YAML. That's why we check
both [Link]('on') and [Link](True).

2 isinstance checks at every level

YAML is flexible; any key could be a scalar, list, or mapping. Defensive isinstance checks prevent AttributeError
crashes on malformed files.

3 Action pinning check

Unpinned GitHub Actions (actions/checkout@v3) can change upstream. Security best practice is to pin to a full
SHA (actions/checkout@abc123).

4 needs reference validation

Cross-job dependencies declared in needs: must reference valid job IDs. Typos here cause workflow failures at
runtime — better to catch statically.

5 Return code for CI

[Link](0 if ok else 1) lets this validator run in a CI pipeline as a pre-commit check.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 52

PART VII — KUBERNETES &

INFRASTRUCTURE (Days 281–330)

Each project in this part is a complete, runnable Python script. Read the description,

1 study the code, then check the line-by-line explanation below it. Type the code
yourself for best retention.
Projects

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 53

DAY 281 ADVANCED

Kubernetes Manifest Validator

Parse and validate Kubernetes YAML manifests for common issues.

■ Source Code
1 import yaml
2 import sys
3 import os
4 from typing import List, Dict
5
6 class K8sValidator:
7 """Validates Kubernetes YAML manifests for correctness and security."""
8
9 VALID_KINDS = {
10 "Pod", "Deployment", "StatefulSet", "DaemonSet", "Job", "CronJob",
11 "Service", "Ingress", "ConfigMap", "Secret", "ServiceAccount",
12 "PersistentVolumeClaim", "Namespace", "NetworkPolicy", "HorizontalPodAutoscaler",
13 }
14
15 def __init__(self):
16 [Link] = []
17 [Link] = []
18
19 def validate_file(self, filepath: str) -> bool:
20 [Link] = []
21 [Link] = []
22
23 with open(filepath) as f:
24 # yaml.safe_load_all handles multi-document YAML (separated by ---)
25 documents = list(yaml.safe_load_all(f))
26
27 for i, doc in enumerate(documents):
28 if doc is None: continue
29 prefix = f"Document {i+1}"
30 self._validate_document(doc, prefix)
31
32 return len([Link]) == 0
33
34 def _validate_document(self, doc: dict, prefix: str):
35 """Validate a single Kubernetes manifest document."""
36 if not isinstance(doc, dict):
37 [Link](f"{prefix}: must be a YAML mapping.")
38 return
39
40 # Every K8s manifest needs apiVersion, kind, and metadata
41 for required_field in ["apiVersion", "kind", "metadata"]:

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 54

42 if required_field not in doc:

43 [Link](f"{prefix}: missing required field '{required_field}'.")
44
45 kind = [Link]("kind", "")
46 if kind and kind not in self.VALID_KINDS:
47 [Link](f"{prefix}: unknown kind '{kind}' (custom resource?).")
48
49 # Validate metadata
50 metadata = [Link]("metadata", {})
51 if isinstance(metadata, dict):
52 self._validate_metadata(metadata, prefix)
53
54 # Kind-specific validation
55 if kind == "Deployment":
56 self._validate_deployment(doc, prefix)
57 elif kind == "Pod":
58 spec = [Link]("spec", {})
59 self._validate_pod_spec(spec, prefix)
60 elif kind == "Service":
61 self._validate_service(doc, prefix)
62 elif kind == "Secret":
63 self._validate_secret(doc, prefix)
64
65 def _validate_metadata(self, metadata: dict, prefix: str):
66 if "name" not in metadata:
67 [Link](f"{prefix}: [Link] is required.")
68
69 # Warn about missing labels
70 labels = [Link]("labels", {})
71 if not labels:
72 [Link](f"{prefix}: no labels defined. Labels are essential for selection.")
73 else:
74 for recommended in ["app", "version"]:
75 if recommended not in labels:
76 [Link](f"{prefix}: recommended label '{recommended}' is missing.")
77
78 def _validate_deployment(self, doc: dict, prefix: str):
79 spec = [Link]("spec", {})
80 if not isinstance(spec, dict):
81 return
82
83 replicas = [Link]("replicas", 1)
84 if replicas < 2:
85 [Link](f"{prefix}: replicas={replicas}. Use >= 2 for HA.")
86
87 # Validate selector matches template labels
88 selector = [Link]("selector", {}).get("matchLabels", {})
89 template_labels = [Link]("template", {}).get("metadata", {}).get("labels", {})
90 for key, value in [Link]():
91 if template_labels.get(key) != value:

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 55

92 [Link](
93 f"{prefix}: [Link].{key}={value} doesn't match template labels."
94 )
95
96 # Validate pod template spec
97 pod_spec = [Link]("template", {}).get("spec", {})
98 self._validate_pod_spec(pod_spec, f"{prefix}/template")
99
100 def _validate_pod_spec(self, spec: dict, prefix: str):
101 if not isinstance(spec, dict): return
102
103 containers = [Link]("containers", [])
104 if not containers:
105 [Link](f"{prefix}: [Link] is empty.")
106 return
107
108 for i, container in enumerate(containers):
109 cpfx = f"{prefix}/containers[{i}]"
110
111 # Image tag pinning
112 image = [Link]("image", "")
113 if [Link](":latest") or ":" not in image:
114 [Link](f"{cpfx}: image '{image}' uses latest tag. Pin to specific version.")
115
116 # Resource limits
117 resources = [Link]("resources", {})
118 if not resources:
119 [Link](f"{cpfx}: no resource limits defined. Risk of OOM kills.")
120 else:
121 if "limits" not in resources:
122 [Link](f"{cpfx}: [Link] not set.")
123 if "requests" not in resources:
124 [Link](f"{cpfx}: [Link] not set.")
125
126 # Security context
127 sc = [Link]("securityContext", {})
128 if not sc:
129 [Link](f"{cpfx}: no securityContext defined.")
130 else:
131 if [Link]("privileged") is True:
132 [Link](f"{cpfx}: privileged=true is a critical security risk.")
133 if [Link]("runAsRoot") is True or [Link]("runAsUser") == 0:
134 [Link](f"{cpfx}: running as root. Set runAsNonRoot: true.")
135
136 def _validate_service(self, doc: dict, prefix: str):
137 spec = [Link]("spec", {})
138 svc_type = [Link]("type", "ClusterIP")
139 if svc_type == "NodePort":
140 [Link](f"{prefix}: NodePort exposes service on all nodes. Use LoadBalancer or
Ingress.")

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 56

141 if svc_type == "LoadBalancer":

142 [Link](f"{prefix}: LoadBalancer provisions a cloud LB (costs money). Verify
intentional.")
143
144 def _validate_secret(self, doc: dict, prefix: str):
145 data = [Link]("data", {})
146 # Secrets should use 'data' (base64) not 'stringData' (plain text in YAML)
147 string_data = [Link]("stringData", {})
148 if string_data:
149 [Link](f"{prefix}: stringData contains plaintext values. They will appear in
YAML files and git history.")
150
151 def report(self):
152 for w in [Link]: print(w)
153 for e in [Link]: print(e)
154 print(f"\nResult: {len([Link])} errors, {len([Link])} warnings")
155
156 if __name__ == "__main__":
157 path = [Link][1] if len([Link]) > 1 else "[Link]"
158 v = K8sValidator()
159 ok = v.validate_file(path)
160 [Link]()
161 [Link](0 if ok else 1)

■ Line-by-Line Explanation

1 yaml.safe_load_all()
Handles multi-document YAML files (documents separated by ---). Returns a generator; wrap in list() to iterate
multiple times.

2 Deployment selector validation

The [Link] must match the pod template labels exactly. A mismatch means the Deployment
can't find its own pods — a common but hard-to-debug mistake.

3 Resource limits required

Without CPU/memory limits, a runaway container can starve other pods. This is one of the most impactful K8s best
practices.

4 privileged: true detection

Privileged containers have full access to the host kernel. This is equivalent to running as root on the host node — a
critical security violation.

5 stringData warning
Secrets in stringData appear in plaintext in YAML files and git history. Use data (base64) and seal secrets with
tools like Sealed Secrets or Vault.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 57

PART VIII — MONITORING &

OBSERVABILITY (Days 331–365)

Each project in this part is a complete, runnable Python script. Read the description,

2 study the code, then check the line-by-line explanation below it. Type the code
yourself for best retention.
Projects

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 58

DAY 331 ADVANCED

Prometheus Metrics Exporter

Build a custom Prometheus exporter that exposes system metrics in the /metrics format.

■ Source Code
1 import [Link] # Built-in HTTP server
2 import time
3 import os
4 import threading
5 import socket
6
7 class MetricsRegistry:
8 """Simple Prometheus-compatible metrics registry."""
9
10 def __init__(self):
11 self._gauges = {} # Current value metrics
12 self._counters = {} # Monotonically increasing metrics
13 self._lock = [Link]() # Thread-safe writes
14
15 def set_gauge(self, name: str, value: float, labels: dict = None, help_text: str = ""):
16 """Set a gauge metric (e.g., current CPU usage)."""
17 with self._lock:
18 key = (name, tuple(sorted((labels or {}).items())))
19 self._gauges[key] = (value, labels or {}, help_text)
20
21 def inc_counter(self, name: str, amount: float = 1.0,
22 labels: dict = None, help_text: str = ""):
23 """Increment a counter metric (e.g., total HTTP requests)."""
24 with self._lock:
25 key = (name, tuple(sorted((labels or {}).items())))
26 current = self._counters.get(key, (0.0, labels or {}, help_text))
27 self._counters[key] = (current[0] + amount, labels or {}, help_text)
28
29 def render(self) -> str:
30 """Render all metrics in Prometheus text exposition format."""
31 lines = []
32
33 with self._lock:
34 # Output HELP and TYPE lines, then metric lines
35 seen_names = set()
36 for (name, label_items), (value, labels, help_text) in self._gauges.items():
37 if name not in seen_names:
38 if help_text:
39 [Link](f"# HELP {name} {help_text}")
40 [Link](f"# TYPE {name} gauge")
41 seen_names.add(name)

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 59

42 label_str = self._format_labels(labels)
43 [Link](f"{name}{label_str} {value}")
44
45 seen_names = set()
46 for (name, label_items), (value, labels, help_text) in self._counters.items():
47 if name not in seen_names:
48 if help_text:
49 [Link](f"# HELP {name} {help_text}")
50 [Link](f"# TYPE {name} counter")
51 seen_names.add(name)
52 label_str = self._format_labels(labels)
53 [Link](f"{name}_total{label_str} {value}")
54
55 return "\n".join(lines) + "\n"
56
57 @staticmethod
58 def _format_labels(labels: dict) -> str:
59 """Format labels as {key="val",key2="val2"}."""
60 if not labels: return ""
61 parts = [f'{k}="{v}"' for k, v in sorted([Link]())]
62 return "{" + ",".join(parts) + "}"
63
64 class SystemCollector:
65 """Collects system metrics and populates the registry."""
66
67 def __init__(self, registry: MetricsRegistry):
68 [Link] = registry
69 [Link] = [Link]()
70
71 def collect(self):
72 """Collect all system metrics."""
73 self._collect_cpu()
74 self._collect_memory()
75 self._collect_disk()
76 self._collect_load()
77
78 def _collect_cpu(self):
79 try:
80 with open("/proc/stat") as f:
81 fields = [Link]().split()
82 total = sum(int(x) for x in fields[1:])
83 idle = int(fields[4])
84 cpu_pct = (total - idle) / total * 100
85 [Link].set_gauge(
86 "node_cpu_usage_percent", round(cpu_pct, 2),
87 labels={"host": [Link]},
88 help_text="CPU usage percentage"
89 )
90 except Exception: pass
91

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 60

92 def _collect_memory(self):
93 try:
94 mem = {}
95 with open("/proc/meminfo") as f:
96 for line in f:
97 k, v = [Link](":", 1)
98 mem[[Link]()] = int([Link]().split()[0]) * 1024
99
100 total = [Link]("MemTotal", 0)
101 available = [Link]("MemAvailable", 0)
102 used = total - available
103
104 [Link].set_gauge("node_memory_total_bytes", total,
105 labels={"host": [Link]}, help_text="Total memory bytes")
106 [Link].set_gauge("node_memory_used_bytes", used,
107 labels={"host": [Link]}, help_text="Used memory bytes")
108 [Link].set_gauge("node_memory_available_bytes", available,
109 labels={"host": [Link]}, help_text="Available memory bytes")
110 except Exception: pass
111
112 def _collect_disk(self):
113 try:
114 st = [Link]("/")
115 total = st.f_blocks * st.f_frsize
116 free = st.f_bavail * st.f_frsize
117 used = total - free
118 [Link].set_gauge("node_disk_total_bytes", total,
119 labels={"host": [Link], "mount": "/"},
120 help_text="Disk total bytes")
121 [Link].set_gauge("node_disk_used_bytes", used,
122 labels={"host": [Link], "mount": "/"},
123 help_text="Disk used bytes")
124 except Exception: pass
125
126 def _collect_load(self):
127 try:
128 load1, load5, load15 = [Link]()
129 [Link].set_gauge("node_load1", load1,
130 labels={"host": [Link]}, help_text="1-minute load average")
131 [Link].set_gauge("node_load5", load5,
132 labels={"host": [Link]}, help_text="5-minute load average")
133 except Exception: pass
134
135 registry = MetricsRegistry()
136 collector = SystemCollector(registry)
137
138 class MetricsHandler([Link]):
139 """HTTP handler that serves Prometheus metrics."""
140
141 def do_GET(self):

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 61

142 if [Link] == "/metrics":

143 [Link]() # Refresh metrics on each scrape
144 body = [Link]().encode("utf-8")
145 self.send_response(200)
146 # Prometheus requires this specific Content-Type
147 self.send_header("Content-Type", "text/plain; version=0.0.4; charset=utf-8")
148 self.send_header("Content-Length", str(len(body)))
149 self.end_headers()
150 [Link](body)
151 elif [Link] == "/health":
152 self.send_response(200)
153 self.end_headers()
154 [Link](b"OK")
155 else:
156 self.send_response(404)
157 self.end_headers()
158
159 def log_message(self, format, *args):
160 # Suppress default access log to keep output clean
161 pass
162
163 if __name__ == "__main__":
164 port = int([Link]("METRICS_PORT", "9100"))
165 server = [Link](("", port), MetricsHandler)
166 print(f"Prometheus exporter running on :{port}/metrics")
167 server.serve_forever()

■ Line-by-Line Explanation

1 Prometheus text format

Each metric block has # HELP (description), # TYPE (gauge/counter/histogram), then metric lines with optional
labels in curly braces.

2 [Link]()
The HTTP server and collector run in the same process. Lock prevents race conditions when the HTTP handler
reads metrics while the collector is writing.

3 Labels in Prometheus
Labels add dimensions to metrics. node_cpu_usage_percent{host='web1'} and
node_cpu_usage_percent{host='web2'} are distinct time series.

4 Collect on scrape
Rather than pushing metrics on a timer, Prometheus pulls metrics by HTTP request. Collecting on demand ensures
freshness without wasted work.

5 Content-Type header
Prometheus requires the specific header text/plain; version=0.0.4 to parse the exposition format correctly. Wrong
Content-Type = parsing failure.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 62

DAY 365 ADVANCED

Full DevOps Pipeline Orchestrator

A complete pipeline runner: lint → test → build → deploy → verify with rollback on failure.

■ Source Code
1 import subprocess
2 import sys
3 import time
4 import os
5 import json
6 from dataclasses import dataclass, field
7 from typing import List, Callable, Optional
8 from enum import Enum
9
10 class StageStatus(Enum):
11 PENDING = "PENDING"
12 RUNNING = "RUNNING"
13 PASSED = "PASSED"
14 FAILED = "FAILED"
15 SKIPPED = "SKIPPED"
16
17 @dataclass
18 class Stage:
19 name: str
20 command: List[str] # Command to run as a list (subprocess-safe)
21 depends_on: List[str] = field(default_factory=list)
22 on_failure: str = "halt" # "halt" | "continue" | "rollback"
23 timeout: int = 300 # Seconds before kill
24 env: dict = field(default_factory=dict) # Extra env vars for this stage
25 status: StageStatus = [Link]
26 duration: float = 0.0
27 output: str = ""
28
29 class Pipeline:
30 """DAG-based pipeline runner with dependency resolution."""
31
32 def __init__(self, name: str):
33 [Link] = name
34 [Link]: List[Stage] = []
35 self.rollback_stages: List[Stage] = []
36 self.start_time = None
37
38 def add_stage(self, stage: Stage) -> "Pipeline":
39 """Add a stage (fluent builder pattern)."""
40 [Link](stage)
41 return self

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 63

42
43 def add_rollback(self, stage: Stage) -> "Pipeline":
44 """Add a rollback stage (run on failure)."""
45 self.rollback_stages.append(stage)
46 return self
47
48 def _resolve_order(self) -> List[Stage]:
49 """Topological sort of stages based on depends_on."""
50 completed = set()
51 ordered = []
52 remaining = list([Link])
53 max_iterations = len(remaining) ** 2 # Detect cycles
54
55 iteration = 0
56 while remaining:
57 iteration += 1
58 if iteration > max_iterations:
59 raise RuntimeError("Circular dependency detected in pipeline stages.")
60
61 for stage in remaining[:]:
62 # A stage is ready if all its dependencies are completed
63 if all(dep in completed for dep in stage.depends_on):
64 [Link](stage)
65 [Link]([Link])
66 [Link](stage)
67
68 return ordered
69
70 def _run_stage(self, stage: Stage) -> bool:
71 """Execute a single stage. Returns True on success."""
72 [Link] = [Link]
73 print(f" ■■ [{[Link]}] Starting...", flush=True)
74
75 start = [Link]()
76
77 # Merge base environment with stage-specific vars
78 env = {**[Link], **[Link]}
79
80 try:
81 result = [Link](
82 [Link],
83 capture_output=True, # Capture stdout and stderr
84 text=True, # Decode bytes to str automatically
85 timeout=[Link],
86 env=env,
87 )
88 [Link] = [Link]() - start
89 [Link] = [Link] + [Link]
90
91 if [Link] == 0:

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 64

92 [Link] = [Link]
93 print(f" ■ [{[Link]}] PASSED ({[Link]:.1f}s)")
94 return True
95 else:
96 [Link] = [Link]
97 print(f" ■ [{[Link]}] FAILED (exit code {[Link]}, {[Link]:.1f}s)")
98 if [Link]():
99 # Print last 5 lines of output for context
100 last_lines = "\n".join([Link]().splitlines()[-5:])
101 print(f" ■ Output (last 5 lines):\n{last_lines}")
102 return False
103
104 except [Link]:
105 [Link] = [Link]
106 [Link] = [Link]
107 print(f" ■ [{[Link]}] TIMEOUT after {[Link]}s")
108 return False
109
110 except FileNotFoundError:
111 [Link] = [Link]
112 print(f" ■ [{[Link]}] FAILED: command not found: {[Link][0]}")
113 return False
114
115 def _run_rollback(self):
116 """Execute rollback stages in reverse order."""
117 if not self.rollback_stages:
118 print(" [No rollback stages defined]")
119 return
120 print("\n === EXECUTING ROLLBACK ===")
121 for stage in reversed(self.rollback_stages):
122 self._run_stage(stage)
123
124 def run(self) -> bool:
125 """Execute the full pipeline. Returns True if all stages pass."""
126 self.start_time = [Link]()
127 print(f"\n■■■■ Pipeline: {[Link]} ■■■")
128 print(f"■ Started: {[Link]('%Y-%m-%d %H:%M:%S')}")
129 print(f"■{'■'*50}")
130
131 try:
132 ordered = self._resolve_order()
133 except RuntimeError as e:
134 print(f"■ ERROR: {e}")
135 return False
136
137 pipeline_success = True
138
139 for stage in ordered:
140 # Skip stages whose dependencies failed
141 if any(

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 65

142 [Link] == [Link]

143 for s in [Link]
144 if [Link] in stage.depends_on
145 ):
146 [Link] = [Link]
147 print(f" ■■ [{[Link]}] SKIPPED (dependency failed)")
148 continue
149
150 success = self._run_stage(stage)
151
152 if not success:
153 pipeline_success = False
154 if stage.on_failure == "rollback":
155 self._run_rollback()
156 break
157 elif stage.on_failure == "halt":
158 print(f" ■ Halting pipeline on failure.")
159 break
160 # "continue": keep going despite failure
161
162 elapsed = [Link]() - self.start_time
163 status_str = "SUCCESS" if pipeline_success else "FAILED"
164 print(f"■{'■'*50}")
165 print(f"■ Status: {status_str} | Duration: {elapsed:.1f}s")
166 print(f"■{'■'*50}\n")
167
168 self._write_report(pipeline_success, elapsed)
169 return pipeline_success
170
171 def _write_report(self, success: bool, elapsed: float):
172 """Write a JSON report for downstream consumption."""
173 report = {
174 "pipeline": [Link],
175 "status": "success" if success else "failed",
176 "duration_seconds": round(elapsed, 2),
177 "timestamp": [Link]("%Y-%m-%dT%H:%M:%SZ", [Link]()),
178 "stages": [
179 {
180 "name": [Link],
181 "status": [Link],
182 "duration": round([Link], 2),
183 }
184 for s in [Link]
185 ]
186 }
187 with open("pipeline_report.json", "w") as f:
188 [Link](report, f, indent=2)
189
190 if __name__ == "__main__":
191 pipeline = Pipeline("my-app-deploy")

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 66

192
193 pipeline.add_stage(Stage("lint", ["python3", "-m", "py_compile", "[Link]"],
194 on_failure="halt"))
195 pipeline.add_stage(Stage("test", ["python3", "-m", "pytest", "tests/", "-q"],
196 depends_on=["lint"], on_failure="halt"))
197 pipeline.add_stage(Stage("build", ["docker", "build", "-t", "my-app:latest", "."],
198 depends_on=["test"], on_failure="halt"))
199 pipeline.add_stage(Stage("deploy", ["kubectl", "apply", "-f", "k8s/"],
200 depends_on=["build"], on_failure="rollback", timeout=120))
201 pipeline.add_stage(Stage("verify", ["curl", "-f", "[Link]
202 depends_on=["deploy"], on_failure="rollback"))
203
204 # Rollback: remove the deployment if verify fails
205 pipeline.add_rollback(Stage("rollback-deploy",
206 ["kubectl", "rollout", "undo", "deployment/my-app"]))
207
208 ok = [Link]()
209 [Link](0 if ok else 1)

■ Line-by-Line Explanation

1 Topological sort (DAG)

Pipelines are directed acyclic graphs where stages have dependencies. The sort ensures dependencies always
run before their dependents.

2 [Link](capture_output=True, text=True)
capture_output=True is shorthand for stdout=PIPE, stderr=PIPE. text=True auto-decodes bytes to str. Always
prefer this over [Link]().

3 [Link]
If a stage exceeds its timeout, the process is killed and this exception is raised. Essential for CI/CD to prevent stuck
pipelines.

4 Rollback stages in reverse order

Deployment rollback is the reverse of deployment. Reversing the rollback stage list mirrors the undo order of the
deployment steps.

5 JSON report output

Machine-readable reports let downstream tools (dashboards, Slack bots, JIRA integrations) consume pipeline
results without parsing human-readable text.

6 on_failure strategies
'halt' stops immediately (fail-fast), 'continue' runs all stages regardless, 'rollback' triggers the undo sequence.
These mirror real CD pipeline semantics.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 67

Complete Project Listing — Days 6 to 365

The following pages contain the complete listing of all 365 projects with descriptions, key concepts, and
implementation notes. The full source code for every project is available in the companion GitHub repository.
Each entry below gives you enough context to implement the project independently.

Part I — Python Foundations

00 Command-Line Argument Parser Beginner

6 Build a reusable CLI parser using argparse — the foundation of every DevOps script.

00 Text File Search Tool Beginner

7 Grep-like tool: search files for patterns using [Link]() with context lines.

00 CSV Report Generator Beginner

8 Read raw data CSVs and produce formatted summary reports with totals and averages.

00 Simple Task Runner Beginner

9 Execute shell commands from Python using [Link]() safely.

01 Password Generator Beginner

0 Generate cryptographically secure passwords using the secrets module.

01 File Watcher Beginner

1 Monitor a directory for changes using a polling loop and [Link]().

01 Disk Usage Analyzer Beginner

2 Find the top 10 largest files in a directory tree using [Link]().

01 Process Manager Beginner

3 List running processes from /proc and filter by name/CPU usage.

01 Cron Expression Parser Beginner

4 Parse cron schedules and calculate next N execution times.

01 INI Config File Handler Beginner

5 Read and write .ini configuration files using the configparser module.

01 Simple HTTP Server Beginner

6 Serve files from a directory using [Link] with custom MIME types.

01 Port Scanner Beginner

7 Scan a host for open ports using socket with threading for speed.

01 JSON Log Formatter Beginner

8 Convert Python logging output to structured JSON for log aggregators.

01 Recursive File Copier Beginner

9 Copy directory trees with progress reporting and conflict resolution.

02 Text Template Engine Beginner

0 Fill in text templates using Python [Link] and custom delimiters.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 68

02 System Info Reporter Beginner

1 Collect and display comprehensive system info: CPU, RAM, disk, network.

02 Uptime Calculator Beginner

2 Read /proc/uptime and format as days/hours/minutes/seconds.

02 Network Interface Lister Beginner

3 Parse /proc/net/if_inet6 and /proc/net/dev for interface statistics.

02 Simple Key-Value Store Beginner

4 Implement a persistent key-value store backed by a JSON file.

02 URL Validator Beginner

5 Validate URLs using [Link] and optional HTTP reachability check.

02 SSH Config Parser Beginner

6 Parse ~/.ssh/config files and extract host aliases, keys, and options.

02 Hosts File Manager Beginner

7 Read, add, and remove entries from /etc/hosts safely.

02 Checksum Verifier Beginner

8 Compute and verify MD5/SHA256 checksums for files using hashlib.

02 Log Rotation Script Beginner

9 Rotate, compress, and archive log files when they exceed a size threshold.

03 Crontab Manager Beginner

0 Read, add, and remove cron jobs from crontab using subprocess.

Part II — Files, Data & Automation

03 Terraform State Parser Intermediate

4 Parse [Link] JSON to extract resource info and dependencies.

03 Ansible Inventory Generator Intermediate

5 Generate Ansible inventory YAML from a CSV or CMDB input.

03 Config Diff Tool Intermediate

6 Compare two YAML/JSON configs and report added, removed, changed keys.

03 Secret Scanner Intermediate

7 Scan source code files for accidentally committed secrets using regex patterns.

03 Markdown to HTML Converter Intermediate

8 Convert Markdown docs to styled HTML using regex transformations.

03 CSV Data Validator Intermediate

9 Validate CSV rows against a schema (types, ranges, regex patterns).

04 Excel Report Generator Intermediate

0 Generate Excel reports with charts using openpyxl.

04 Database Migration Script Intermediate

1 Apply SQL migration files in order with up/down rollback support.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 69

04 SQLite Query Tool Intermediate

2 Interactive SQLite query runner with output formatting and CSV export.

04 File Deduplication Tool Intermediate

3 Find and remove duplicate files using content hashing.

04 Archive Extractor Intermediate

4 Auto-detect and extract .[Link], .zip, .bz2 archives.

04 Config Merger Tool Intermediate

5 Deep-merge multiple YAML config files with priority ordering.

04 Service Dependency Graph Intermediate

6 Build a dependency graph from [Link] and detect cycles.

04 Prometheus Alert Rule Generator Intermediate

7 Generate PromQL alert rules YAML from a high-level spec.

04 [Link] Auditor Intermediate

8 Check Python dependencies for outdated versions and known CVEs.

04 SSL Certificate Scanner Intermediate

9 Batch-check SSL certificates for expiry across a list of hostnames.

05 Docker Compose Validator Intermediate

0 Validate [Link] files for syntax and best practices.

05 Logstash Config Generator Intermediate

1 Generate Logstash pipeline configs from a template spec.

05 Release Notes Generator Intermediate

2 Parse git log --oneline and generate structured release notes.

05 README Generator Intermediate

3 Auto-generate project [Link] from code comments and metadata.

05 License Header Checker Intermediate

4 Verify that all source files have the required license header.

05 API Versioning Manager Intermediate

5 Track and validate semantic version bumps in API changelogs.

05 Infrastructure Cost Estimator Intermediate

6 Estimate cloud costs from a resource list using pricing APIs.

05 Slack Message Sender Intermediate

7 Send formatted Slack notifications via Incoming Webhooks.

05 PagerDuty Alert Creator Intermediate

8 Create and resolve PagerDuty incidents via the REST API.

05 JIRA Ticket Creator Intermediate

9 Programmatically create JIRA issues from a Python script.

06 Email Notifier Intermediate

0 Send HTML email notifications using smtplib and MIME types.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 70

06 Webhook Receiver Intermediate

1 Handle incoming webhooks (GitHub, GitLab) and trigger actions.

06 Rate Limiter Implementation Intermediate

2 Implement token bucket and sliding window rate limiters.

06 Task Queue (Redis-backed) Intermediate

3 Simple job queue using Redis lists for worker-based task processing.

06 Config Encryption Tool Intermediate

4 Encrypt/decrypt sensitive config values using Fernet symmetric encryption.

06 Audit Log Writer Intermediate

5 Write tamper-evident audit logs with HMAC signatures.

06 Deployment Manifest Builder Intermediate

6 Generate K8s deployment manifests from a simple spec file.

06 Health Check Aggregator Intermediate

7 Aggregate health checks from multiple services into a dashboard.

06 SLA Calculator Intermediate

8 Calculate SLA compliance from incident log data.

06 Capacity Planner Intermediate

9 Project future resource needs from historical trend data.

07 Service Mesh Config Generator Intermediate

0 Generate Istio/Linkerd service mesh configs from specs.

07 Alertmanager Route Generator Intermediate

1 Generate Prometheus Alertmanager routing YAML configs.

07 Grafana Dashboard Exporter Intermediate

2 Export Grafana dashboard JSON via API and save to files.

07 Log Anomaly Detector Intermediate

3 Detect anomalous log patterns using rolling statistics.

07 Retry Decorator Library Intermediate

4 Build a @retry decorator with configurable backoff strategies.

07 CLI Progress Bar Intermediate

5 Implement a terminal progress bar from scratch.

07 Parallel File Processor Intermediate

6 Process files in parallel using [Link].

07 Workflow State Machine Intermediate

7 Implement a state machine for deployment workflow management.

07 Inventory Reconciler Intermediate

8 Compare actual vs desired infrastructure state and report diffs.

07 Change Request Generator Intermediate

9 Auto-generate change request documents from deployment plans.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 71

08 On-Call Schedule Manager Intermediate

0 Parse and manage on-call rotations from a YAML schedule.

Part III — Networking & APIs

08 DNS Resolver Tool Intermediate

3 Resolve DNS records (A, AAAA, MX, TXT) using [Link]().

08 TCP Port Forwarder Intermediate

4 Forward TCP connections between hosts using threading and socket.

08 Network Bandwidth Monitor Intermediate

5 Read /proc/net/dev periodically to calculate interface throughput.

08 HTTP Load Generator Intermediate

6 Generate HTTP load with configurable RPS using threading.

08 Webhook Relay Server Intermediate

7 Receive webhooks and relay them to multiple downstream endpoints.

08 OAuth2 Client Intermediate

8 Implement the OAuth2 client credentials flow for API authentication.

08 GraphQL Client Intermediate

9 Query GraphQL APIs using urllib with introspection support.

09 gRPC Health Checker Intermediate

0 Check gRPC service health using the standard health checking protocol.

09 Service Mesh Topology Mapper Intermediate

1 Discover and map microservice dependencies via traffic analysis.

09 TLS Certificate Generator Intermediate

2 Generate self-signed TLS certificates using the cryptography library.

09 Nginx Config Generator Intermediate

3 Generate NGINX server block configs from a deployment spec.

09 HAProxy Config Builder Intermediate

4 Build HAProxy frontend/backend configs from a service inventory.

09 VPN Status Monitor Intermediate

5 Monitor OpenVPN/WireGuard tunnel status and peer connectivity.

09 IP Reputation Checker Intermediate

6 Check IPs against threat intelligence feeds (AbuseIPDB API).

09 Network Topology Scanner Intermediate

7 Discover hosts on a subnet using ICMP ping sweeps.

09 Load Balancer Health Checker Intermediate

8 Validate load balancer backends and report unhealthy nodes.

09 Reverse Proxy Tester Intermediate

9 Verify correct routing behaviour through a reverse proxy.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 72

10 CORS Policy Validator Intermediate

0 Test CORS headers on API endpoints for security compliance.

10 API Gateway Stub Intermediate

1 Build a minimal API gateway with routing, auth, and rate limiting.

10 Service Registry Client Intermediate

2 Register and discover services using Consul's HTTP API.

10 Circuit Breaker Implementation Advanced

3 Implement the circuit breaker pattern for resilient service calls.

10 Distributed Tracing Emitter Advanced

4 Emit OpenTelemetry traces from a Python service.

10 Message Queue Producer/Consumer Advanced

5 Produce and consume messages with RabbitMQ via pika.

10 Kafka Producer/Consumer Advanced

6 Stream events with Apache Kafka using kafka-python.

10 Redis Pub/Sub System Advanced

7 Implement a pub/sub message bus using Redis channels.

10 gRPC Service Implementation Advanced

8 Build a gRPC server and client with Protocol Buffers.

10 WebSocket Server Advanced

9 Real-time bidirectional communication using websockets library.

11 Service Discovery Client Advanced

0 Auto-discover services using DNS-SD / mDNS.

11 Network Policy Enforcer Advanced

1 Validate and enforce network security policies as code.

11 Zero-Trust Authenticator Advanced

2 Implement mTLS mutual authentication between services.

11 API Fuzzer Advanced

3 Automated API endpoint fuzzing for security testing.

11 HTTP/2 Client Advanced

4 Make HTTP/2 requests with server push support using h2 library.

11 CDN Cache Purger Advanced

5 Purge CloudFront/Fastly cache entries via API.

11 DNS-over-HTTPS Client Advanced

6 Query DNS using HTTPS (DoH) for privacy and filtering.

11 SMTP Server Stub Advanced

7 Minimal SMTP server for testing email sending in development.

11 Network Packet Analyser Advanced

8 Capture and decode network packets using raw sockets.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 73

11 BGP Route Monitor Advanced

9 Monitor BGP routing table changes via exabgp API.

12 Anycast Health Probe Advanced

0 Probe anycast services from multiple vantage points.

12 Network Chaos Injector Advanced

1 Inject network faults (latency, packet loss) using tc/netem.

12 Service Latency Profiler Advanced

2 Measure p50/p95/p99 latencies across a microservice graph.

12 API Contract Tester Advanced

3 Validate API responses against OpenAPI schemas.

12 Synthetic Monitor Advanced

4 Simulate user journeys via HTTP sequences and alert on failure.

12 TCP Connection Pool Advanced

5 Implement a thread-safe TCP connection pool for reuse.

12 HTTP/3 QUIC Probe Advanced

6 Test QUIC/HTTP3 availability on target endpoints.

12 Service Level Objective Tracker Advanced

7 Track SLO compliance in real time with burn rate alerts.

12 IPAM Database Manager Advanced

8 Manage IP address allocations in a SQLite IPAM database.

12 Network Configuration Backup Advanced

9 SSH into network devices and backup running configurations.

13 Full Observability Agent Advanced

0 Combined metrics + logs + traces collector with OTLP export.

Part IV — Docker & Containers

13 Docker Image Builder Intermediate

3 Programmatically build Docker images using the Docker API.

13 Container Log Streamer Intermediate

4 Stream container logs in real time via the Docker API.

13 Docker Network Manager Intermediate

5 Create, list, and connect containers to Docker networks.

13 Container Resource Limiter Intermediate

6 Apply CPU/memory limits to running containers via Docker API.

13 Docker Volume Manager Intermediate

7 Create, inspect, and clean up Docker volumes programmatically.

13 Registry Image Scanner Intermediate

8 List and analyse images in a private Docker registry.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 74

13 Container Health Monitor Intermediate

9 Monitor container health status and restart unhealthy ones.

14 Docker Compose Orchestrator Intermediate

0 Start, stop, and manage multi-container apps with compose.

14 Image Layer Analyser Intermediate

1 Inspect Docker image layers and measure size contributions.

14 Container Escape Detector Advanced

2 Detect common container escape vulnerabilities in images.

14 OCI Image Builder Advanced

3 Build OCI-compliant container images without Docker.

14 Container Networking Deep Dive Advanced

4 Inspect container network namespaces and virtual ethernet pairs.

14 Rootless Container Runner Advanced

5 Run containers without root using user namespace remapping.

14 CGroups Resource Monitor Advanced

6 Read cgroup v2 files to monitor container resource usage.

14 eBPF Container Tracer Advanced

7 Use BCC/eBPF to trace container system calls.

14 Container Immutability Enforcer Advanced

8 Validate that containers run with read-only root filesystems.

14 Distroless Image Validator Advanced

9 Verify images use distroless or scratch base with no shell.

15 SBOM Generator Advanced

0 Generate Software Bill of Materials from container images.

15 Container Signing Tool Advanced

1 Sign container images with Sigstore/cosign.

15 Runtime Security Monitor Advanced

2 Detect anomalous container behaviour at runtime with Falco rules.

15 Image Vulnerability Scanner Advanced

3 Scan images for CVEs using Trivy API integration.

15 Docker Swarm Manager Advanced

4 Deploy and scale services on Docker Swarm clusters.

15 Container Registry Mirror Advanced

5 Implement a pull-through cache for a container registry.

15 Buildkit Remote Builder Advanced

6 Trigger BuildKit remote builds and capture artefacts.

15 Multi-Arch Image Builder Advanced

7 Build multi-architecture images using QEMU emulation.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 75

15 Container Diff Tool Advanced

8 Compare two container images and report filesystem differences.

15 Image Promotion Pipeline Advanced

9 Promote images between registries after security gates pass.

16 Kata Container Launcher Advanced

0 Launch lightweight VM-isolated containers with Kata.

16 WebAssembly Runtime Advanced

1 Run WASM modules in Python using wasmtime bindings.

16 Container Chaos Tool Advanced

2 Randomly stop/pause containers to test system resilience.

16 Sidecar Injector Advanced

3 Programmatically inject sidecar containers into pod specs.

16 Container to Serverless Migrator Advanced

4 Analyse container manifests and generate Lambda/Cloud Run equivalents.

16 Container Capacity Planner Advanced

5 Project cluster capacity needs from resource usage trends.

16 Container Drift Detector Advanced

6 Detect runtime filesystem drift from the image baseline.

16 Privileged Container Auditor Advanced

7 Audit running containers for security violations.

16 Container Startup Profiler Advanced

8 Profile container startup time and identify slow init steps.

16 Image Cache Warmer Advanced

9 Pre-pull required images on cluster nodes before deployments.

17 Container Forensics Tool Advanced

0 Collect forensic artefacts from stopped containers.

17 CNI Plugin Tester Advanced

1 Test CNI network plugin configuration and behaviour.

17 Container Registry Replication Advanced

2 Replicate images between geographically distributed registries.

17 Init Container Generator Advanced

3 Generate init container specs for dependency management.

17 Container Log Aggregator Advanced

4 Collect logs from all containers and ship to a central store.

17 Docker Bench Security Runner Advanced

5 Automate Docker CIS benchmark security checks.

17 Container Network Policy Generator Advanced

6 Generate K8s NetworkPolicies from container communication patterns.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 76

17 Ephemeral Container Launcher Advanced

7 Launch debug containers into running pods for troubleshooting.

17 Container Audit Trail Advanced

8 Record all container lifecycle events to an audit database.

17 Registry Garbage Collector Advanced

9 Identify and delete unreferenced image layers in a registry.

18 Container Platform Health Dashboard Advanced

0 Aggregate health metrics across all container infrastructure.

Part V — AWS & Cloud

18 IAM Permission Analyser Intermediate

3 List IAM users, roles, and policies; flag overpermissioned entities.

18 CloudWatch Log Streamer Intermediate

4 Stream CloudWatch log groups in real time using the Boto3 API.

18 RDS Snapshot Manager Intermediate

5 List, create, and delete RDS snapshots with retention policy.

18 Lambda Function Deployer Intermediate

6 Package and deploy AWS Lambda functions from Python scripts.

18 CloudFormation Stack Manager Intermediate

7 Deploy, update, and delete CloudFormation stacks.

18 Route53 DNS Manager Intermediate

8 Manage DNS records in Route 53 hosted zones.

18 ECS Task Runner Intermediate

9 Run one-off ECS Fargate tasks programmatically.

19 SQS Queue Monitor Intermediate

0 Poll SQS queue depth and dead-letter queue metrics.

19 SNS Notification Publisher Intermediate

1 Publish messages to SNS topics with structured payloads.

19 DynamoDB CRUD Tool Intermediate

2 Full CRUD operations on DynamoDB tables with pagination.

19 CloudTrail Audit Analyser Intermediate

3 Parse CloudTrail logs to audit API call patterns.

19 Cost Explorer Reporter Intermediate

4 Fetch and report AWS costs by service, region, and tag.

19 AWS Config Compliance Checker Intermediate

5 Check resource compliance against Config rules.

19 Parameter Store Manager Intermediate

6 Store, retrieve, and rotate secrets in Systems Manager Parameter Store.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 77

19 Secrets Manager Rotator Intermediate

7 Implement automatic secret rotation in Secrets Manager.

19 VPC Flow Log Analyser Intermediate

8 Parse VPC Flow Logs to detect unusual traffic patterns.

19 Security Group Auditor Intermediate

9 Find security groups with overly permissive inbound rules.

20 Auto Scaling Group Manager Intermediate

0 Monitor and adjust ASG desired capacity based on custom metrics.

20 EKS Cluster Manager Advanced

1 Manage EKS clusters: create nodegroups, update configs, scale.

20 ECR Image Lifecycle Manager Advanced

2 Implement lifecycle policies to clean old ECR images.

20 CloudFront Cache Manager Advanced

3 Manage CloudFront distributions and cache invalidations.

20 AWS Organisations Policy Manager Advanced

4 Apply Service Control Policies across organisational units.

20 Cross-Account Role Assumer Advanced

5 Assume IAM roles across AWS accounts for cross-account automation.

20 AWS Backup Orchestrator Advanced

6 Manage backup plans, vaults, and restore testing.

20 GuardDuty Finding Processor Advanced

7 Process GuardDuty findings and route to incident response workflows.

20 AWS WAF Rule Manager Advanced

8 Manage WAF web ACLs and IP set rules programmatically.

20 EventBridge Rule Creator Advanced

9 Create event-driven automation rules with EventBridge.

21 Step Functions Workflow Builder Advanced

0 Define and execute Step Functions state machines from Python.

21 Glue ETL Job Manager Advanced

1 Trigger and monitor AWS Glue ETL jobs via Python.

21 Athena Query Runner Advanced

2 Execute Athena SQL queries and download results to S3.

21 Kinesis Stream Producer Advanced

3 Produce high-throughput event streams to Kinesis Data Streams.

21 MSK Kafka Manager Advanced

4 Manage Managed Streaming for Kafka clusters via API.

21 AppMesh Service Mesh Manager Advanced

5 Configure AWS App Mesh virtual services and routers.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 78

21 Transit Gateway Route Manager Advanced

6 Manage Transit Gateway route tables across VPCs.

21 Direct Connect Monitor Advanced

7 Monitor Direct Connect virtual interfaces and BGP sessions.

21 AWS SSO Session Manager Advanced

8 Automate AWS SSO authentication and credential vending.

21 Control Tower Account Vending Advanced

9 Automate new account creation via Control Tower.

22 Well-Architected Review Tool Advanced

0 Automate AWS Well-Architected Framework reviews via API.

22 Trusted Advisor Analyser Advanced

1 Parse Trusted Advisor checks and generate action plans.

22 AWS Budgets Alerting System Advanced

2 Create and monitor Budget alerts with SNS notifications.

22 S3 Intelligent Tiering Migrator Advanced

3 Migrate objects to intelligent tiering to reduce costs.

22 Multi-Region Failover Controller Advanced

4 Automate Route53 health check-based regional failover.

22 AWS Resource Tagger Advanced

5 Apply consistent tags across all untagged resources.

22 Config Drift Detector Advanced

6 Detect infrastructure drift between IaC definitions and actual state.

22 AWS Incident Response Runbook Advanced

7 Automate incident response actions using SSM automation.

22 Service Quota Monitor Advanced

8 Track AWS service quota usage and alert before limits are hit.

22 Landing Zone Builder Advanced

9 Bootstrap a new AWS landing zone with baseline controls.

23 Cloud Governance Reporter Advanced

0 Generate a comprehensive cloud governance compliance report.

Part VI — CI/CD & Automation

23 Git Hook Manager Intermediate

2 Install and manage pre-commit, pre-push git hooks.

23 Commit Message Validator Intermediate

3 Enforce Conventional Commits format in git commit messages.

23 Branch Protection Checker Intermediate

4 Verify branch protection rules via GitHub/GitLab API.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 79

23 Pull Request Automator Intermediate

5 Auto-label, assign, and comment on PRs based on changed files.

23 Release Tag Manager Intermediate

6 Automate semantic version tag creation and release notes.

23 CI Pipeline Reporter Intermediate

7 Collect CI/CD pipeline metrics and generate trend reports.

23 Test Coverage Enforcer Intermediate

8 Fail the pipeline if code coverage drops below a threshold.

23 Artifact Uploader Intermediate

9 Upload build artifacts to S3/Nexus/Artifactory after builds.

24 Dependency Graph Generator Intermediate

0 Map inter-service dependencies from IaC and manifests.

24 Deployment Frequency Tracker Intermediate

1 Measure DORA deployment frequency from CI/CD event logs.

24 MTTR Calculator Intermediate

2 Calculate Mean Time to Recovery from incident tickets.

24 Change Failure Rate Tracker Intermediate

3 Track percentage of deployments that cause incidents.

24 Lead Time Calculator Intermediate

4 Measure commit-to-production lead time from git and CD logs.

24 Feature Flag Manager Intermediate

5 Manage feature flags with environment-based overrides.

24 Blue-Green Deployment Script Advanced

6 Orchestrate blue-green deployments with traffic switching.

24 Canary Release Controller Advanced

7 Gradually shift traffic to new versions with automatic rollback.

24 GitOps Reconciler Advanced

8 Reconcile Git-defined desired state with actual cluster state.

24 ArgoCD Application Manager Advanced

9 Manage ArgoCD applications and sync status via API.

25 Flux Reconciler Monitor Advanced

0 Monitor FluxCD reconciliation events and alert on failures.

25 Helm Chart Tester Advanced

1 Render and validate Helm charts with multiple value sets.

25 Kustomize Overlay Builder Advanced

2 Generate environment-specific Kustomize overlays.

25 Policy as Code Engine Advanced

3 Enforce OPA/Rego policies on Kubernetes manifests in CI.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 80

25 Security Scan Aggregator Advanced

4 Aggregate SAST, DAST, and SCA results into a unified report.

25 Container Image Scan Gate Advanced

5 Block deployments if image vulnerability score exceeds threshold.

25 SBOM Compliance Checker Advanced

6 Verify software supply chain compliance from SBOM files.

25 Sigstore Verification Tool Advanced

7 Verify container image signatures using Sigstore policy.

25 Secrets Rotation Pipeline Advanced

8 Automate secret rotation across services without downtime.

25 Ephemeral Environment Manager Advanced

9 Create and destroy per-PR preview environments automatically.

26 Multi-Cluster Deployment Tool Advanced

0 Deploy applications across multiple K8s clusters simultaneously.

26 Rollback Automation Tool Advanced

1 Automatically roll back deployments on SLO breach.

26 Post-Deployment Smoke Tester Advanced

2 Run smoke tests after every deployment and alert on failure.

26 Infrastructure Test Runner Advanced

3 Run Terratest/pytest-testinfra tests against infrastructure.

26 Compliance Pipeline Gate Advanced

4 Block deployments that violate security compliance policies.

26 Dependency Update Bot Advanced

5 Automatically raise PRs to update outdated dependencies.

26 Code Quality Gate Advanced

6 Enforce code quality metrics (complexity, duplication) in CI.

26 E2E Test Orchestrator Advanced

7 Schedule and parallelise end-to-end test suites.

26 Performance Regression Detector Advanced

8 Compare benchmark results across commits to detect regressions.

26 Database Schema Migration CI Advanced

9 Validate and apply database migrations in CI/CD pipelines.

27 API Mock Server Generator Advanced

0 Generate WireMock stubs from OpenAPI specs for integration testing.

27 Chaos Engineering Pipeline Advanced

1 Inject chaos experiments as automated pipeline stages.

27 Multi-Cloud Deployment Router Advanced

2 Route deployments to AWS/GCP/Azure based on policy.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 81

27 Environment Cleanup Scheduler Advanced

3 Automatically clean up ephemeral test environments.

27 Deployment Risk Scorer Advanced

4 Score deployment risk based on code change size and test coverage.

27 CI/CD Cost Optimiser Advanced

5 Identify and eliminate wasteful CI/CD resource usage.

27 Supply Chain Security Scanner Advanced

6 Audit the software supply chain from source to production.

27 Progressive Delivery Controller Advanced

7 Orchestrate Argo Rollouts progressive delivery strategies.

27 Multi-Tenancy Deployment Manager Advanced

8 Deploy the same app to multiple tenants with isolated configs.

27 Golden Path Template Generator Advanced

9 Generate project scaffolding from golden path templates.

28 DevSecOps Policy Enforcer Advanced

0 Enforce security policies at every stage of the pipeline.

Part VII — Kubernetes & Infrastructure

28 Pod Resource Right-Sizer Advanced

2 Analyse actual vs requested resources and recommend adjustments.

28 Kubernetes RBAC Auditor Advanced

3 Audit K8s RBAC permissions for excessive privilege.

28 Namespace Resource Quota Manager Advanced

4 Manage resource quotas and LimitRanges across namespaces.

28 HPA Tuner Advanced

5 Automatically tune HorizontalPodAutoscaler settings from metrics.

28 Node Drain Orchestrator Advanced

6 Safely drain K8s nodes for maintenance with PDB respect.

28 Pod Disruption Budget Manager Advanced

7 Create and validate PodDisruptionBudgets for all workloads.

28 K8s Event Monitor Advanced

8 Watch Kubernetes events and alert on Warning events.

28 ConfigMap/Secret Syncer Advanced

9 Sync ConfigMaps and Secrets from Vault or Parameter Store.

29 Custom Resource Definition Manager Advanced

0 Create and manage CRDs with validation and conversion webhooks.

29 Kubernetes Operator Framework Advanced

1 Build a minimal Kubernetes operator with reconciliation loop.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 82

29 Admission Webhook Server Advanced

2 Implement a validating/mutating admission webhook in Python.

29 K8s Audit Log Analyser Advanced

3 Parse K8s audit logs to detect suspicious API activity.

29 Multi-Cluster Config Manager Advanced

4 Manage kubeconfig contexts and switch clusters safely.

29 Namespace Cost Allocator Advanced

5 Allocate cluster costs to namespaces based on resource usage.

29 K8s Garbage Collector Advanced

6 Find and clean up orphaned K8s resources.

29 Pod Security Policy Migrator Advanced

7 Migrate from PSP to Pod Security Admission (PSA).

29 Cert-Manager Certificate Reporter Advanced

8 Report on certificate expiry across a K8s cluster.

29 Velero Backup Manager Advanced

9 Schedule and monitor Velero cluster backups.

30 K8s Ingress Manager Advanced

0 Manage Ingress objects and validate routing rules.

30 Service Account Token Auditor Advanced

1 Audit long-lived service account tokens for security compliance.

30 Helm Release Manager Advanced

2 Manage Helm releases with upgrade, rollback, and diff.

30 K8s Network Policy Validator Advanced

3 Validate that NetworkPolicies provide expected isolation.

30 Cluster Upgrade Planner Advanced

4 Assess K8s version compatibility before cluster upgrades.

30 Node Affinity Manager Advanced

5 Apply node affinity/anti-affinity rules to workloads.

30 Kubernetes Dashboard API Client Advanced

6 Query the K8s metrics server for dashboard data.

30 Custom Scheduler Plugin Advanced

7 Implement a custom K8s scheduler extender plugin.

30 Volume Snapshot Manager Advanced

8 Create and restore persistent volume snapshots.

30 StatefulSet Migration Tool Advanced

9 Migrate stateful workloads between storage classes.

31 K8s Config Drift Detector Advanced

0 Detect configuration drift between GitOps repo and live cluster.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 83

31 Terraform Module Validator Advanced

1 Validate Terraform modules against naming and structure conventions.

31 Pulumi Stack Manager Advanced

2 Manage Pulumi stacks and track resource outputs.

31 Ansible Playbook Tester Advanced

3 Test Ansible playbooks against Molecule test instances.

31 Chef/Puppet Compliance Scanner Advanced

4 Scan managed nodes for policy compliance.

31 Infrastructure Dependency Map Advanced

5 Build a visual map of infrastructure resource dependencies.

31 Terraform Drift Reporter Advanced

6 Run terraform plan and report on infrastructure drift.

31 IaC Cost Estimator Advanced

7 Estimate costs for Terraform plans before apply.

31 Infrastructure Inventory Advanced

8 Build a complete inventory of all cloud and on-prem resources.

31 Compliance as Code Engine Advanced

9 Evaluate infrastructure against CIS benchmark controls.

32 Multi-Cloud Resource Manager Advanced

0 Unified interface to manage resources across AWS, GCP, Azure.

32 Infrastructure Change Approver Advanced

1 Enforce approval workflows for high-risk infrastructure changes.

32 Disaster Recovery Orchestrator Advanced

2 Automate failover and failback procedures for DR.

32 Service Catalogue Manager Advanced

3 Manage a self-service infrastructure service catalogue.

32 Infrastructure as Data Pipeline Advanced

4 Export all infrastructure config to a queryable data store.

32 GitOps Policy Engine Advanced

5 Enforce GitOps policies: no manual changes, all via PRs.

32 Platform Engineering CLI Advanced

6 Build an internal developer platform CLI for self-service.

32 Infrastructure Scoring Tool Advanced

7 Score infrastructure maturity against the DORA framework.

32 Chaos Engineering Framework Advanced

8 Build a chaos engineering test framework for infrastructure.

32 Multi-Region Replication Manager Advanced

9 Replicate data and config across multiple cloud regions.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 84

33 Infrastructure Security Posture Manager Advanced

0 Continuously assess and report infrastructure security posture.

Part VIII — Monitoring & Observability

33 OpenTelemetry Tracer Advanced

2 Instrument a Python service with OTLP traces and spans.

33 Loki Log Pusher Advanced

3 Push structured logs to Grafana Loki via the HTTP API.

33 Jaeger Trace Viewer Advanced

4 Query and visualise distributed traces from Jaeger API.

33 Alertmanager Webhook Handler Advanced

5 Receive and process Prometheus Alertmanager webhook payloads.

33 SLO Error Budget Calculator Advanced

6 Calculate error budgets from SLI metrics and alert on burn rate.

33 Incident Timeline Builder Advanced

7 Build a timeline of events from multiple data sources for post-mortems.

33 MTTR/MTTD Dashboard Advanced

8 Track and visualise incident response time metrics.

33 Log Pattern Miner Advanced

9 Mine recurring log patterns to identify operational noise.

34 Anomaly Detection Engine Advanced

0 Detect metric anomalies using rolling z-scores.

34 On-Call Rotation Manager Advanced

1 Manage on-call schedules and escalation policies.

34 Post-Mortem Report Generator Advanced

2 Generate structured post-mortem reports from incident data.

34 Service Dependency Health Map Advanced

3 Real-time health map of all services and their dependencies.

34 Capacity Forecasting Tool Advanced

4 Forecast future resource needs using linear regression.

34 FinOps Dashboard Generator Advanced

5 Generate cloud cost dashboards from billing API data.

34 Chaos Monkey Implementation Advanced

6 Randomly terminate instances to test resilience.

34 Observability Pipeline Advanced

7 Route, filter, and enrich telemetry data before export.

34 SRE Runbook Automator Advanced

8 Automate common SRE runbook actions via triggered scripts.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 85

34 Platform SLO Reporter Advanced

9 Generate weekly SLO compliance reports for stakeholders.

35 Incident Response Bot Advanced

0 Slack bot that triages incidents and runs diagnostic commands.

35 Full-Stack APM Agent Advanced

1 Collect application performance metrics across the full stack.

35 Real User Monitoring Collector Advanced

2 Collect and analyse real user performance data.

35 Synthetic Transaction Monitor Advanced

3 Run synthetic user transactions and alert on degradation.

35 Log-Based Alerting Engine Advanced

4 Alert on log patterns without a time-series database.

35 Metric Correlation Engine Advanced

5 Correlate metrics across services to identify root causes.

35 Distributed Profiler Advanced

6 Continuous profiling of Python services with py-spy.

35 Alert Fatigue Reducer Advanced

7 Deduplicate and suppress noisy alerts intelligently.

35 Observability as Code Tool Advanced

8 Define dashboards, alerts, and runbooks as Python code.

35 SLI Measurement Framework Advanced

9 Instrument services to measure SLIs against SLO targets.

36 Telemetry Gateway Advanced

0 Unified gateway that accepts metrics, logs, and traces.

36 AI-Assisted Root Cause Analyser Advanced

1 Use LLM API to correlate symptoms and suggest root causes.

36 Predictive Scaling Engine Advanced

2 Predict traffic spikes and pre-scale infrastructure.

36 Observability Maturity Assessor Advanced

3 Assess and score observability maturity against a rubric.

36 Platform Engineering Dashboard Advanced

4 Unified engineering efficiency and health dashboard.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis

365 Python Projects for DevOps Engineers Page 86

Keep Building. Keep Shipping.

Every expert DevOps engineer was once a beginner who kept writing one more script.

[Link]/akashfrancis3211 | [Link]/in/akash-francis91

[Link]/akashfrancis3211 | [Link]/in/akash-francis91 | © 2025 Akash Francis