Python模組化設計與資料函式庫操作結合應用

# calculator.py
def add(x, y):
 # 加法運算函式
 return x + y

def subtract(x, y):
 # 減法運算函式
 return x - y

def multiply(x, y):
 # 乘法運算函式
 return x * y

def divide(x, y):
 # 除法運算函式，包含除零錯誤處理
 if y == 0:
 raise ZeroDivisionError("無法除以零")
 return x / y

# main.py
import calculator

def main():
 try:
 print("加法運算結果：", calculator.add(10, 5))
 print("減法運算結果：", calculator.subtract(10, 5))
 print("乘法運算結果：", calculator.multiply(10, 5))
 print("除法運算結果：", calculator.divide(10, 2))
 # 測試除零錯誤
 print("除法運算結果：", calculator.divide(10, 0))
 except ZeroDivisionError as e:
 print("錯誤訊息：", e)

if __name__ == "__main__":
 main()

#!/usr/bin/python3
import sys
import os
import urllib.request

def main():
 # 使用sys模組取得Python版本資訊
 print("Python版本：{}.{}.{}".format(*sys.version_info))
 
 # 使用os模組取得作業系統資訊
 print("作業系統型別：", os.name)
 print("環境變數PATH：", os.getenv('PATH'))
 print("目前工作目錄：", os.getcwd())

 # 使用urllib.request模組傳送網路請求
 try:
 with urllib.request.urlopen('http://example.com') as response:
 page = response.read()
 print("網頁內容：", str(page, encoding='utf-8'))
 except urllib.error.URLError as e:
 print("網路請求錯誤：", e)

if __name__ == "__main__":
 main()

#!/usr/bin/python3
from mysql.connector import MySQLConnection, Error
from config import read_mysql_config

def update_image(author_id, filename):
 # 從組態檔案讀取資料函式庫連線引數
 config = read_mysql_config()
 query = 'UPDATE authors SET photo = %s WHERE id = %s'
 
 try:
 # 建立資料函式庫連線
 connection = MySQLConnection(**config)
 cursor = connection.cursor()
 
 # 讀取圖片檔案
 with open(filename, 'rb') as file:
 photo = file.read()
 
 # 執行更新操作
 cursor.execute(query, (photo, author_id))
 connection.commit()
 print("圖片更新成功")
 
 except Error as e:
 print("資料函式庫操作錯誤：", e)
 
 finally:
 # 確保連線被關閉
 if connection.is_connected():
 connection.close()

def main():
 author_id = 47
 image_path = "/home/user/images/author.jpg"
 update_image(author_id, image_path)

if __name__ == "__main__":
 main()

#!/usr/bin/python3
from mysql.connector import MySQLConnection, Error
from config import read_mysql_config

def read_image(author_id, output_filename):
 config = read_mysql_config()
 query = 'SELECT photo FROM authors WHERE id = %s'
 
 try:
 connection = MySQLConnection(**config)
 cursor = connection.cursor()
 cursor.execute(query, (author_id,))
 
 # 取得圖片資料
 photo = cursor.fetchone()[0]
 
 # 將圖片寫入檔案
 with open(output_filename, 'wb') as file:
 file.write(photo)
 print("圖片儲存成功")
 
 except Error as e:
 print("資料函式庫操作錯誤：", e)
 
 finally:
 if connection.is_connected():
 connection.close()

def main():
 author_id = 47
 output_path = "/home/user/images/output.jpg"
 read_image(author_id, output_path)

if __name__ == "__main__":
 main()

# date_time_utils.py
import sys
import datetime

def get_python_version():
 return f"Python版本：{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}"

def get_current_time():
 current_time = datetime.datetime.now()
 return current_time.strftime("%Y-%m-%d %H:%M:%S")

# test_date_time_utils.py
import unittest
from date_time_utils import get_python_version, get_current_time

class TestDateTimeUtils(unittest.TestCase):
 def test_python_version_format(self):
 version = get_python_version()
 self.assertRegex(version, r'^Python版本：\d+\.\d+\.\d+$')

 def test_current_time_format(self):
 current_time = get_current_time()
 self.assertRegex(current_time, r'^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$')

if __name__ == "__main__":
 unittest.main()

import socket

def get_ip_address(domain):
 try:
 ip_address = socket.gethostbyname(domain)
 return ip_address
 except socket.gaierror:
 return "無法解析網域名稱"

def main():
 domain = "www.example.com"
 print(f"{domain} 的IP位址是：{get_ip_address(domain)}")

if __name__ == "__main__":
 main()

import nmap

def scan_network(target):
 nm = nmap.PortScanner()
 nm.scan(target, '1-1024')

 for host in nm.all_hosts():
 print(f"主機：{host}")
 for proto in nm[host].all_protocols():
 print(f"協定：{proto}")
 ports = nm[host][proto].keys()
 for port in sorted(ports):
 state = nm[host][proto][port]['state']
 print(f"埠號：{port} 狀態：{state}")

def main():
 target = '127.0.0.1'
 scan_network(target)

if __name__ == "__main__":
 main()

# 安裝 Docker
sudo apt-get update
sudo apt-get install docker.io
# 啟動 Docker 服務
sudo systemctl start docker
sudo systemctl enable docker
# 驗證安裝結果
docker --version

import os
import yaml
from kubernetes import client, config

# 載入 Kubernetes 組態
config.load_kube_config()

# 建立 Kubernetes API 客戶端
v1 = client.CoreV1Api()

def create_test_job(test_config):
    # 建立測試任務的組態
    job_config = {
        'apiVersion': 'batch/v1',
        'kind': 'Job',
        'metadata': {'name': test_config['job_name']},
        'spec': {
            'completions': test_config['parallelism'],
            'template': {
                'spec': {
                    'containers': [{
                        'name': 'test-container',
                        'image': test_config['image'],
                        'command': test_config['command']
                    }],
                    'restartPolicy': 'Never'
                }
            }
        }
    }
    
    # 建立 Job 物件
    batch_v1 = client.BatchV1Api()
    batch_v1.create_namespaced_job(
        namespace="default",
        body=job_config
    )

# 範例組態
test_config = {
    'job_name': 'example-test',
    'parallelism': 3,
    'image': 'test-image:latest',
    'command': ['python', 'run_tests.py']
}

# 建立測試任務
create_test_job(test_config)

import pandas as pd

class TestDataManager:
    def __init__(self, data_path):
        self.data_path = data_path
        self.data = self.load_data()
    
    def load_data(self):
        return pd.read_csv(self.data_path)
    
    def get_test_cases(self, condition):
        return self.data.query(condition)

# 使用範例
data_manager = TestDataManager('test_data.csv')
specific_tests = data_manager.get_test_cases("priority == 'high'")

def scale_test_environment(load):
    if load > 0.8:
        # 增加資源
        scale_up()
    elif load < 0.2:
        # 減少資源
        scale_down()

def scale_up():
    # 實作擴充套件邏輯
    pass

def scale_down():
    # 實作縮減邏輯
    pass