Module 7

01-python-foundations/my_system_health.py

@@ -0,0 +1,32 @@
+import psutil
+
+try:
+    threshold = input("Enter the CPU threshold (%): ")
+    threshold = float(threshold)
+except ValueError:
+    print("That's not a valid number. Please enter something like 75 or 80.5")
+    exit()
+
+cpu_usage = psutil.cpu_percent(interval=1)
+memory_usage = psutil.virtual_memory().percent
+disk_usage = psutil.disk_usage("/").percent
+
+print("You entered:", threshold)
+print("Current CPU usage:", cpu_usage)
+print("Current Memory usage:", memory_usage)
+print("Current Disk usage:", disk_usage)
+
+if cpu_usage > threshold:
+    print("CPU status: WARNING - usage is above threshold")
+else:
+    print("CPU status: Healthy")
+
+if memory_usage > threshold:
+    print("Memory status: WARNING - usage is above threshold")
+else:
+    print("Memory status: Healthy")
+
+if disk_usage > threshold:
+    print("Disk status: WARNING - usage is above threshold")
+else:
+    print("Disk status: Healthy")

02-apis-and-json/call_api.py

@@ -2,7 +2,7 @@
 
 import requests
 
-API_URL = "https://jsonplaceholder.typicode.com/todos/1"
+API_URL = "https://jsonplaceholder.typicode.com/todos/2"
 
 
 def fetch_todo(url):
@@ -17,8 +17,8 @@ def main():
     for key, value in todo.items():
         print(f"{key:10}: {value}")
 
-    if todo.get("userId") == 1:
-        print("\n>> This todo belongs to user 1")
+    if todo.get("userId") == 2:
+        print("\n>> This todo belongs to user 2")
 
 
 if __name__ == "__main__":

02-apis-and-json/github_user.py

@@ -0,0 +1,24 @@
+import json
+import requests
+
+username = input("Enter a GitHub username: ")
+url = f"https://api.github.com/users/{username}"
+
+try:
+    response = requests.get(url, timeout=10)
+    response.raise_for_status()
+except requests.exceptions.HTTPError:
+    print(f"Could not find a GitHub user called '{username}'.")
+    exit()
+
+data = response.json()
+
+print("Name        :", data.get("name"))
+print("Public repos:", data.get("public_repos"))
+print("Followers   :", data.get("followers"))
+print("Location    :", data.get("location"))
+
+with open("github_user.json", "w", encoding="utf-8") as f:
+    json.dump(data, f, indent=2)
+
+print("\nSaved full response to github_user.json")

02-apis-and-json/stock_market_api.py

@@ -24,7 +24,7 @@ def get_daily_series(symbol, api_key):
 def main():
     api_key = os.environ.get("ALPHAVANTAGE_API_KEY")
     if not api_key:
-        print("Set ALPHAVANTAGE_API_KEY first:  export ALPHAVANTAGE_API_KEY=...")
+        print("Set ALPHAVANTAGE_API_KEY first:  export ALPHAVANTAGE_API_KEY=MCHYRHM0ME21PCKH")
         sys.exit(1)
 
     symbol = input("Enter a stock symbol (e.g. IBM, AMZN, GOOGL): ").strip().upper()

03-file-handling-and-logs/my_log_analyzer.py

@@ -0,0 +1,40 @@
+import json
+
+LOG_FILE = "app.log"
+
+try:
+    with open(LOG_FILE, "r", encoding="utf-8") as f:
+        lines = f.readlines()
+except FileNotFoundError:
+    print(f"Log file not found: {LOG_FILE}")
+    exit()
+
+print("Total lines:", len(lines))
+
+info_count = 0
+warning_count = 0
+error_count = 0
+
+for line in lines:
+    words = line.split()
+    if "INFO" in words:
+        info_count += 1
+    if "WARNING" in words:
+        warning_count += 1
+    if "ERROR" in words:
+        error_count += 1
+
+print("INFO   :", info_count)
+print("WARNING:", warning_count)
+print("ERROR  :", error_count)
+
+summary = {
+    "INFO": info_count,
+    "WARNING": warning_count,
+    "ERROR": error_count,
+}
+
+with open("log_summary.json", "w", encoding="utf-8") as f:
+    json.dump(summary, f, indent=2)
+
+print("\nSaved summary to log_summary.json")

04-object-oriented-python/log_analyzer_oop.py

@@ -9,7 +9,12 @@
 class LogAnalyzer:
     def __init__(self, log_file):
         self.log_file = log_file
-        self.counts = {level: 0 for level in LEVELS}
+        self.counts = {
+            "INFO": 0,
+            "WARNING": 0,
+            "ERROR": 0,
+            "UNKNOWN": 0,
+        }
 
     def read_logs(self):
         try:
@@ -32,7 +37,9 @@ def analyze(self, lines):
                 self.counts["UNKNOWN"] += 1
         return self.counts
 
-    def write_summary(self, path="log_counts.json"):
+    def write_summary(self, path=None):
+        if path is None:
+            path = Path(self.log_file).with_suffix(".json")
         with open(path, "w", encoding="utf-8") as f:
             json.dump(self.counts, f, indent=2)
 
@@ -47,6 +54,8 @@ def main():
         return
 
     result = analyzer.analyze(lines)
+    analyzer.write_summary("custom_log_counts.json")
+
     print("Log Analysis Summary:")
     for level, count in result.items():
         print(f"  {level:7}: {count}")

05-cli-tools-argparse/my_log_analyzer_cli.py

@@ -0,0 +1,52 @@
+import argparse
+import json
+import sys
+from pathlib import Path
+
+LEVELS = ("INFO", "WARNING", "ERROR")
+
+parser = argparse.ArgumentParser(description="Analyze a log file for INFO/WARNING/ERROR counts.")
+parser.add_argument("--file", required=True, help="path to the log file")
+parser.add_argument("--out", help="write the summary to this JSON file")
+parser.add_argument("--level", choices=LEVELS, help="show the count for only this level")
+
+args = parser.parse_args()
+
+log_path = Path(args.file)
+if not log_path.is_file():
+    print(f"Error: log file not found: {args.file}", file=sys.stderr)
+    sys.exit(2)
+
+with open(log_path, "r", encoding="utf-8") as f:
+    lines = f.readlines()
+
+info_count = 0
+warning_count = 0
+error_count = 0
+
+for line in lines:
+    words = line.split()
+    if "INFO" in words:
+        info_count += 1
+    if "WARNING" in words:
+        warning_count += 1
+    if "ERROR" in words:
+        error_count += 1
+
+counts = {
+    "INFO": info_count,
+    "WARNING": warning_count,
+    "ERROR": error_count,
+}
+
+if args.level:
+    print(f"{args.level}: {counts[args.level]}")
+else:
+    print("INFO   :", info_count)
+    print("WARNING:", warning_count)
+    print("ERROR  :", error_count)
+
+if args.out:
+    with open(args.out, "w", encoding="utf-8") as f:
+        json.dump(counts, f, indent=2)
+    print(f"Wrote summary to {args.out}")

05-cli-tools-argparse/my_summary.json

@@ -0,0 +1,5 @@
+{
+  "INFO": 10,
+  "WARNING": 2,
+  "ERROR": 3
+}

05-cli-tools-argparse/summary.json

@@ -0,0 +1,5 @@
+{
+  "INFO": 10,
+  "WARNING": 2,
+  "ERROR": 3
+}

07-apis-with-fastapi/devops-utilities-api/app/api.py

@@ -19,6 +19,9 @@ def hello():
 def health():
     return {"status": "ok"}
 
+@app.get("/version")
+def version():
+    return {"version": "1.2.0"}
 
 app.include_router(metrics.router)
 app.include_router(logs.router)

07-apis-with-fastapi/devops-utilities-api/routers/logs.py

@@ -12,3 +12,23 @@ def get_log_summary(file: str | None = None):
         return analyze_logs(file)
     except FileNotFoundError:
         raise HTTPException(status_code=404, detail=f"Log file not found: {file}")
+
+
+@router.get("/logs/errors", status_code=200)
+def get_log_errors(file: str | None = None):
+    """Return only the ERROR count from the log file."""
+    try:
+        summary = analyze_logs(file)
+        return {"log_file": summary["log_file"], "error_count": summary["counts"]["ERROR"]}
+    except FileNotFoundError:
+        raise HTTPException(status_code=404, detail=f"Log file not found: {file}")
+
+
+@router.get("/logs/warnings", status_code=200)
+def get_log_warnings(file: str | None = None):
+    """Return only the WARNING count from the log file."""
+    try:
+        summary = analyze_logs(file)
+        return {"log_file": summary["log_file"], "warning_count": summary["counts"]["WARNING"]}
+    except FileNotFoundError:
+        raise HTTPException(status_code=404, detail=f"Log file not found: {file}")

07-apis-with-fastapi/devops-utilities-api/routers/metrics.py

@@ -6,8 +6,8 @@
 
 
 @router.get("/metrics", status_code=200)
-def get_metrics():
+def get_metrics(cpu_threshold: int | None = None):
     try:
-        return get_system_metrics()
+        return get_system_metrics(cpu_threshold)
     except Exception as exc:
         raise HTTPException(status_code=500, detail=f"Could not read metrics: {exc}")

07-apis-with-fastapi/devops-utilities-api/services/metrics_service.py

@@ -1,22 +1,22 @@
 import psutil
 
 
-def get_system_metrics():
+def get_system_metrics(cpu_threshold: int | None = None):
     """Read CPU, memory and disk usage, and flag high CPU against a threshold."""
     cpu_percent = psutil.cpu_percent(interval=1)
     memory_percent = psutil.virtual_memory().percent
     disk_percent = psutil.disk_usage("/").percent
 
-    cpu_threshold = 85
+    if cpu_threshold is None:
+        cpu_threshold = 85
 
     status = "High CPU" if cpu_percent > cpu_threshold else "Healthy"
 
     return {
-        "cpu_percentage":cpu_percent,
-        "memory_percentage":memory_percent,
-        "disk_percentage":disk_percent,
-        "cpu_threshold":cpu_threshold,
-        "system_status":status
+        "cpu_percentage": cpu_percent,
+        "memory_percentage": memory_percent,
+        "disk_percentage": disk_percent,
+        "cpu_threshold": cpu_threshold,
+        "system_status": status
     }
 
-

08-ai-agents-for-devops/README.md

@@ -70,3 +70,4 @@ Wire up the agent yourself:
 2. Pass the `analyze_log_file` tool to `create_agent(model, tools=[...], system_prompt=...)`.
 3. Invoke it with a `HumanMessage` asking it to analyze `app.log`, and print the result.
 4. Bonus: add a second tool (e.g. return the last N lines) and watch how it chooses.
+5. The module now includes a `read_log_tail` tool so the agent can return the last N lines of the log file on request.

08-ai-agents-for-devops/log_utils.py

@@ -20,3 +20,11 @@ def count_log_levels(text):
 
 def read_log_file(path):
     return Path(path).read_text(encoding="utf-8")
+
+
+def read_log_tail(path: str, n: int = 5) -> str:
+    text = read_log_file(path)
+    lines = text.splitlines()
+    if n <= 0:
+        return ""
+    return "\n".join(lines[-n:])

08-ai-agents-for-devops/run_agent.py

@@ -16,14 +16,15 @@
 from langgraph.graph.message import add_messages
 from langgraph.prebuilt import ToolNode, tools_condition
 
-from log_utils import LEVELS, count_log_levels, read_log_file
+from log_utils import LEVELS, count_log_levels, read_log_file, read_log_tail
 
 APP_LOG = str(Path(__file__).parent / "app.log")
 MODEL = os.environ.get("OLLAMA_MODEL", "llama3.2")
 
 SYSTEM_PROMPT = (
     "You are a Log Analysis Agent for DevOps engineers. "
     "Always use the analyze_log_file tool to get exact counts, never guess. "
+    "If the user asks for actual log text, use the read_log_tail tool. "
     "State the INFO/WARNING/ERROR counts, then give a one or two line summary. "
     "Suggest ideas only, never perform production actions."
 )
@@ -36,12 +37,18 @@ def analyze_log_file(path: str) -> str:
     return ", ".join(f"{level}={counts[level]}" for level in LEVELS)
 
 
+@tool
+def read_log_tail_tool(path: str, n: int = 5) -> str:
+    """Return the last n lines of a log file."""
+    return read_log_tail(path, n)
+
+
 def make_model():
     return ChatOllama(model=MODEL, base_url="http://localhost:11434", temperature=0)
 
 
 def build_agent():
-    return create_agent(make_model(), tools=[analyze_log_file], system_prompt=SYSTEM_PROMPT)
+    return create_agent(make_model(), tools=[analyze_log_file, read_log_tail_tool], system_prompt=SYSTEM_PROMPT)
 
 
 # The same agent <-> tools loop that create_agent gives you, built by hand so you
@@ -51,14 +58,14 @@ class AgentState(TypedDict):
 
 
 def build_custom_agent():
-    llm_with_tools = make_model().bind_tools([analyze_log_file])
+    llm_with_tools = make_model().bind_tools([analyze_log_file, read_log_tail_tool])
 
     def call_model(state: AgentState):
-        return {"messages": [llm_with_tools.invoke(state["messages"])]}
+        return {"messages": [llm_with_tools.invoke(state["messages"]) ]}
 
     graph = StateGraph(AgentState)
     graph.add_node("agent", call_model)
-    graph.add_node("tools", ToolNode([analyze_log_file]))
+    graph.add_node("tools", ToolNode([analyze_log_file, read_log_tail_tool]))
     graph.add_edge(START, "agent")
     graph.add_conditional_edges("agent", tools_condition)
     graph.add_edge("tools", "agent")

08-ai-agents-for-devops/test_log_utils.py

@@ -26,3 +26,15 @@ def test_bundled_app_log():
     app_log = Path(__file__).parent / "app.log"
     counts = count_log_levels(read_log_file(str(app_log)))
     assert counts == {"INFO": 10, "WARNING": 2, "ERROR": 3}
+
+
+def test_read_log_tail():
+    from log_utils import read_log_tail
+
+    path = Path(__file__).parent / "app.log"
+    text = read_log_file(str(path))
+    lines = text.splitlines()
+    expected = "\n".join(lines[-2:])
+
+    assert read_log_tail(str(path), n=2) == expected
+    assert read_log_tail(str(path), n=0) == ""