Workshop/python

.gitignore

@@ -43,3 +43,8 @@ override.tf.json
 # Ignore CLI configuration files
 .terraformrc
 terraform.rckeys/
+
+lab2/myapp.log
+
+.venv
+.venv/*

Readme.md

@@ -0,0 +1 @@
+pip install -r requirements.txt

apilab/main.py

@@ -0,0 +1,47 @@
+import httpx
+import json
+import pprint
+import time
+
+
+def main():
+    # Task 1
+    user_id = input("Please type the user's id you want: ")
+    try:
+        response = httpx.get("https://jsonplaceholder.typicode.com/users?id=" + user_id)
+        response.raise_for_status()
+        chosen_user = response.json()[0]
+        print("User name: " + str(chosen_user["name"]) + "\nEmail: " + str(chosen_user["email"]) + "\nAddress: " + str(chosen_user["address"]["street"]) + ", " + str(chosen_user["address"]["city"]) )
+    except (httpx.HTTPStatusError, IndexError):            
+        if response.status_code == 500:
+            print("Server error. Please try again later.")
+        else:
+            print("User not found\n")
+
+    # Task 2
+
+    for i in range(3):
+        try:
+            print("Fetching System metrics ...")
+            URL = "https://api.example.com/system/metrics"
+            headers = {"Authorization": "Bearer YOUR_API_KEY"}
+            params = {"metrics": "cpu,memory"}
+            response = httpx.get(URL, params=params, headers=headers)
+            if response.status_code == 401:
+                print("Invalid API key")
+                break
+            if response.status_code == 500:
+                print("Server is currently down")
+                break
+        except httpx.HTTPError:
+            print(f"attempt {i + 1} failed: server is currently down.")
+            time.sleep(2)
+            print("Retrying in 2 seconds.")
+
+
+
+
+
+
+if __name__=="__main__":
+    main()

boto3lab/main.py

@@ -0,0 +1,74 @@
+import boto3
+import random
+
+# s3_client = boto3.client("s3")
+# response = s3_client.list_buckets()
+
+# for bucket in response["Buckets"]:
+#     print(f"Buckets: {bucket['Name']}")
+def main():
+    while True:
+        user_choice = input("[1] Manage S3 Buckets\n[2] Manage EC2 Instances\n[3] Exit\n")
+        if user_choice == "1":
+            s3_client = boto3.client("s3")
+            user_choice = input("chose an option: list, create, delete: ")
+            if user_choice == "list":
+                list_bucket(s3_client)
+            elif user_choice == "create":
+                create_bucket(s3_client)
+            elif user_choice == "delete":
+                delete_bucket()
+            else:
+                print("Please chose a valid option from the menu")
+        elif user_choice =="2":
+            ec2_client = boto3.client("ec2")
+            user_choice = input("chose an option: list, start, stop, terminate: ")
+            if user_choice == "list":
+                list_instances(ec2_client)
+            elif user_choice == "start":
+                pass
+            elif user_choice == "stop":
+                pass
+            elif user_choice == "terminate":
+                pass
+            else:
+                print("Please chose a valid option from the menu")
+        elif user_choice == "3":
+            break
+        else:
+            print("Please chose a valid option from the menu")
+
+
+
+def list_bucket(s3_client):
+    response = s3_client.list_buckets()
+    try:
+        for bucket in response["Buckets"]:
+            print(f"Buckets: {bucket['Name']}")
+    except KeyError:
+        print("no bucket was found")
+def create_bucket(s3_client):
+    bucket_name = input("Please enter the new bucket name: ")
+    response = s3_client.create_bucket(
+        Bucket=bucket_name
+        )
+    print(bucket_name + " was created")     
+def delete_bucket(s3_client):
+        bucket_name = input("Please type the name of the bucket you want to delete: ")
+        response = s3_client.delete_bucket(
+            Bucket=bucket_name,
+            ExpectedBucketOwner='???'
+
+        )
+def list_instances(ec2_client):
+    response = ec2_client.describe_instances()
+    try:
+        for instance in response["Instances"]:
+            print(f"Instance: {instance['Name']}")
+    except KeyError:
+        print("no instance was found")
+
+
+if __name__ == '__main__': 
+    main() 
+

boto3lab/main2.py

@@ -0,0 +1,68 @@
+import boto3
+import os
+
+def list_bucket(s3_client):
+    response = s3_client.list_buckets()
+    try:
+        for bucket in response["Buckets"]:
+            print(f"Buckets: {bucket['Name']}")
+    except KeyError:
+        print("No buckets were found")
+def create_bucket(s3_client):
+    bucket_name = input("Please chose a name for the new bucket: ")
+    response = s3_client.create_bucket(
+        Bucket=bucket_name
+        )
+    print(bucket_name + " was created")     
+def delete_bucket(s3_client):
+        bucket_name = input("Please type the name of the bucket you want to delete: ")
+        bucket_owner = input("Please type the name of the owner: ")
+        response = s3_client.delete_bucket(
+            Bucket=bucket_name,
+            ExpectedBucketOwner=bucket_owner               
+        )
+def s3_menu():
+        s3_client = boto3.client("s3")
+        while True:
+            print("Welcome to S3 menu. Please choose what you want to do:")
+            user_choice = input("[1] List of all buckets.\n[2] Create a new bucket.\n[3] Delete bucket.\n[4] Go back to main menu.\n")
+            match user_choice:
+                case "1":
+                    list_bucket(s3_client)
+                case "2":
+                    create_bucket(s3_client)
+                case "3":
+                    delete_bucket(s3_client)
+                case "4":
+                    print("Exited to main menu.")
+                    return
+                case default:
+                    print("Invalid option. Please select again.")
+
+
+def ec2_menu():
+    pass
+
+def menu():
+    os.environ['AWS_ACCESS_KEY_ID'] = input("AWS_ACCESS_KEY_ID: ").strip()
+    os.environ['AWS_SECRET_ACCESS_KEY'] = input("AWS_SECRET_ACCESS_KEY: ").strip()
+    os.environ['AWS_DEFAULT_REGION'] = input("AWS_DEFAULT_REGION: ").strip()
+    while True:
+        user_choice = input("Chose an option from the menu:\n[1] Manage S3 Buckets\n[2] Manage EC2 Instances\n[3] Exit\n")
+        match user_choice:
+            case "1":
+                s3_menu()
+            case "2":
+                ec2_menu()
+            case "3":
+                break
+            case default:
+                print("Invalid option. Please select again.")
+
+
+
+
+def main():
+    menu()
+if __name__ == "__main__":
+    main()

lab/invalid_server_error.py

@@ -0,0 +1,2 @@
+class InvalidServerError(Exception):
+    pass

lab/log.py

@@ -0,0 +1,30 @@
+import logging
+import sys
+import json
+import os
+
+class JasonFormatter(logging.Formatter):
+    def format(self, record: logging.LogRecord):
+        log = {
+               "timestamp": self.formatTime(record),
+               "level": record.levelname,
+               "message": record.getMessage()
+               }
+        return json.dumps(log)
+def setup_logging():
+    log_level = os.environ.get("LOG_LEVEL", "DEBUG")
+    log_format = os.environ.get("LOG_FORMAT", "TEXT")
+
+    stdout_handler = logging.StreamHandler(sys.stdout)
+    logger = logging.getLogger("myapp")
+    file_handler = logging.FileHandler("myapp.log")
+    logger.addHandler(stdout_handler)
+    logger.addHandler(file_handler)
+    logger.setLevel(log_level)
+    if log_format == "JSON":
+        stdout_handler.setFormatter(JasonFormatter())
+        file_handler.setFormatter(JasonFormatter())
+    else:
+        stdout_handler.setFormatter(logging.Formatter("%(asctime)s:%(name)s:%(levelname)s:%(module)s:%(lineno)d:%(funcName)s:%(message)s"))
+        file_handler.setFormatter(logging.Formatter("%(asctime)s:%(name)s:%(levelname)s:%(module)s:%(lineno)d:%(funcName)s:%(message)s"))
+    return logger

lab/main.py

@@ -0,0 +1,29 @@
+from fastapi import FastAPI
+import json
+import httpx
+from dataclasses import dataclass
+from datetime import datetime
+import modules
+
+app = FastAPI()
+servers = modules.read_server_list()
+
+@dataclass
+class ServerStatusResponse:
+    server_name: str
+    server_status: str | bool
+
+@app.get("/server")
+def get_server(server_name: str) -> ServerStatusResponse:
+    for server in servers:
+        if server.name == server_name:
+            return ServerStatusResponse(server_name, server.online)
+    return ServerStatusResponse(server_name, server_name + " Does not exist.")
+
+
+@app.post("/server")
+def add_server(server_name: str) -> ServerStatusResponse:
+    new_server = modules.Server(name=server_name, online=True, cpus=6, ram=10)
+    modules.add_new_server(new_server)
+    servers.append(new_server)
+    return ServerStatusResponse(server_name, " Server was added and now is running.")

lab/modules.py

@@ -0,0 +1,36 @@
+from pydantic import BaseModel, ValidationError
+import json
+
+
+class ServerStatusResponse(BaseModel):
+    server_name: str
+    server_status: str | bool
+
+
+class Server(BaseModel):
+    name: str
+    online: bool
+    cpus: int
+    ram: int
+
+
+def read_server_list() -> list[Server]:
+    with open("servers.txt", "r") as f:
+        servers: list[Server] = []
+        for line in f.readlines():
+            if line.strip():
+                json_object = json.loads(line)
+                try:
+                    new_server = Server(**json_object)
+                except ValidationError:
+                    pass
+                else:
+                    servers.append(new_server)
+    return servers
+
+
+def add_new_server(new_server: Server):
+    with open("servers.txt", "a") as f:
+        f.write("\n")
+        f.write(new_server.model_dump_json())
+

lab/myapp.log

@@ -0,0 +1,5 @@
+2025-02-10 15:25:39,050:myapp:INFO:main:57:<module>:Valid Server name.
+2025-02-10 15:25:50,234:myapp:ERROR:main:59:<module>:Invalid Server name.
+2025-02-10 16:01:15,471:myapp:INFO:main:25:<module>:Valid Server name.
+2025-02-10 16:01:19,268:myapp:ERROR:main:27:<module>:Invalid Server name.
+2025-02-10 16:01:23,829:myapp:ERROR:main:27:<module>:Invalid Server name.

lab/servers.txt

@@ -0,0 +1,4 @@
+{"name": "nginx", "online": true, "cpus": 2, "ram": 4}
+{"name":"wow","online":true,"cpus":6,"ram":10}
+{"name":"wow","online":true,"cpus":6,"ram":10}
+{"name":"yonatan","online":true,"cpus":6,"ram":10}

lab/valid_servers.py

@@ -0,0 +1,3 @@
+valid_servers = {
+    "nginx": True, "docker": False
+}

plab/main.py

@@ -0,0 +1,26 @@
+import subprocess
+try:
+    p = subprocess.run(["ls", "/var/log"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+except FileNotFoundError:
+    print("the command does not exist")
+except PermissionError:
+    print("do not have permission to run the process")
+err = p.stderr.decode()
+out = p.stdout.decode()
+if "No such file or directory" in err:
+    print("No such file or directory")
+if "permission" in err:
+    print("permission dined")
+if p.returncode == 0:
+    print(out)
+
+
+p = subprocess.run(["systemctl", "status", "nginx"], stdout=subprocess.PIPE)
+if "inactive" in p.stdout.decode():
+    p = subprocess.run(["systemctl", "restart", "nginx"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    print(p.returncode)
+    if p.returncode == 0:
+        print("restart successful")
+    else:
+        print("restart failed")
+

requirements.txt

@@ -0,0 +1,3 @@
+httpx
+fastapi
+pydantic