Python CLI จัดการ GCS ให้ทนทานและเร็วขึ้น

สวัสดีครับ

วันนี้ผมจะพาทุกคนมาดูวิธีสร้าง Python CLI ที่เอาไว้จัดการไฟล์บน Google Cloud Storage (GCS) ของเรานะครับ

เวลาที่เราต้องอัปโหลดหรือดาวน์โหลดไฟล์เยอะๆ ผ่าน API บางทีมันก็มีปัญหาเน็ตหลุด หรือ API ตอบช้าบ้างอะไรบ้างใช่ไหมครับ

เรามาเพิ่มความทนทาน (resilience) และความเร็ว (optimization) ให้กับมันกันดีกว่าครับ

1. ติดตั้งไลบรารีที่ต้องใช้กันก่อนนะครับ

pip install google-cloud-storage tenacity

2. โค้ดสำหรับอัปโหลดไฟล์แบบพื้นฐาน

แบบนี้ก็ใช้ได้นะครับ แต่ถ้าเน็ตหลุดกลางคัน หรือเซิฟเวอร์ GCS มีปัญหาจิ๊ดๆ มันก็พังเลยนะ

import os
from google.cloud import storage

def upload_blob_basic(bucket_name, source_file_name, destination_blob_name):
    \"\"\"อัปโหลดไฟล์ขึ้น GCS แบบธรรมดาครับ\"\"\"
    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(destination_blob_name)

    blob.upload_from_filename(source_file_name)

    print(f\"File {source_file_name} uploaded to {destination_blob_name} in bucket {bucket_name} นะครับ\")

# ตัวอย่างการใช้งานจริง
# ต้องเปลี่ยน 'your-bucket-name' เป็นชื่อ Bucket ของเพื่อนๆ นะครับ
# และสร้างไฟล์ 'my_test_file.txt' ก่อนรันด้วยนะ
#
# with open(\"my_test_file.txt\", \"w\") as f:
#     f.write(\"Hello GCS from basic upload!\")
# upload_blob_basic(\"your-bucket-name\", \"my_test_file.txt\", \"test_folder/my_test_file.txt\")

3. เพิ่มความทนทานด้วย tenacity (Resilience)

เวลาเน็ตมีปัญหา API ล่มบ้าง เราก็อยากให้มันลองใหม่เองใช่ไหมครับ tenacity ช่วยได้เลยนะ มันจะ Retry ให้เราอัตโนมัติครับ

import os
from google.cloud import storage
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from google.api_core import exceptions as api_exceptions # เอาไว้จับ Error ของ Google API ครับ

@retry(stop=stop_after_attempt(5), # ลองใหม่สูงสุด 5 ครั้งครับ
       wait=wait_exponential(multiplier=1, min=4, max=10), # รอแบบทวีคูณ เริ่ม 4 วิ แล้วก็ 8 วิ ไปเรื่อยๆ แต่ไม่เกิน 10 วิครับ
       retry=retry_if_exception_type((api_exceptions.ServiceUnavailable,
                                      api_exceptions.TooManyRequests,
                                      api_exceptions.InternalServerError,
                                      api_exceptions.GatewayTimeout)), # ระบุ Error ที่อยากให้ Retry ครับ
       reraise=True) # ถ้าลองครบแล้วยังไม่ได้ ให้โยน Error ขึ้นมาครับ
def upload_blob_resilient(bucket_name, source_file_name, destination_blob_name):
    \"\"\"อัปโหลดไฟล์ขึ้น GCS แบบทนทานครับ\"\"\"
    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(destination_blob_name)

    print(f\"กำลังพยายามอัปโหลด {source_file_name} ไปที่ {destination_blob_name} (รอบที่...) ครับ\")
    blob.upload_from_filename(source_file_name)

    print(f\"File {source_file_name} อัปโหลดสำเร็จแล้วนะครับ\")

# ตัวอย่างการใช้งานจริง
# with open(\"my_resilient_file.txt\", \"w\") as f:
#     f.write(\"Content for resilient upload!\")
# upload_blob_resilient(\"your-bucket-name\", \"my_resilient_file.txt\", \"test_folder/my_resilient_file.txt\")

4. เพิ่มความเร็วด้วยการอัปโหลดหลายไฟล์พร้อมกัน (Optimization)

ถ้ามีไฟล์ต้องอัปโหลดเยอะๆ เราไม่ต้องรอทีละไฟล์ก็ได้ครับ ใช้ concurrent.futures ช่วยให้มันวิ่งพร้อมๆ กันเลย ทีนี้งานก็จะเสร็จไวขึ้นเยอะเลยนะ

import os
from google.cloud import storage
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from google.api_core import exceptions as api_exceptions
from concurrent.futures import ThreadPoolExecutor # ใช้สำหรับ multi-threading นะครับ

# เราเอาฟังก์ชันอัปโหลดที่ทนทาน (resilient) มาใช้ต่อได้เลยครับ
# อาจจะต้องปรับนิดหน่อยให้มันรับ bucket_name, source_file_name, destination_blob_name ตรงๆ
@retry(stop=stop_after_attempt(5),
       wait=wait_exponential(multiplier=1, min=4, max=10),
       retry=retry_if_exception_type((api_exceptions.ServiceUnavailable,
                                      api_exceptions.TooManyRequests,
                                      api_exceptions.InternalServerError,
                                      api_exceptions.GatewayTimeout)),
       reraise=True)
def upload_blob_resilient_concurrent_task(bucket_name, source_file_name, destination_blob_name):
    \"\"\"ฟังก์ชันอัปโหลดทนทาน สำหรับใช้ใน ThreadPoolExecutor ครับ\"\"\"
    storage_client = storage.Client() # Client ควรจะสร้างในแต่ละ thread ถ้าเป็นไปได้ หรือส่งมาจากภายนอกให้ปลอดภัย
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(destination_blob_name)
    print(f\"กำลังพยายามอัปโหลด (concurrent) {source_file_name} ไปที่ {destination_blob_name} (รอบที่...) ครับ\")
    blob.upload_from_filename(source_file_name)
    print(f\"File {source_file_name} อัปโหลดสำเร็จ (concurrent) แล้วนะครับ\")

def upload_multiple_files(bucket_name, file_paths, destination_prefix=\"\", max_workers=5):
    \"\"\"อัปโหลดหลายไฟล์พร้อมกันเลยครับ โดยใช้ ThreadPoolExecutor\"\"\"
    print(f\"กำลังจะอัปโหลด {len(file_paths)} ไฟล์พร้อมกัน {max_workers} Workers นะครับ\")
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = []
        for file_path in file_paths:
            destination_blob_name = f\"{destination_prefix}{os.path.basename(file_path)}\"
            futures.append(executor.submit(upload_blob_resilient_concurrent_task,
                                        bucket_name, file_path, destination_blob_name))

        # รอให้ทุกไฟล์อัปโหลดเสร็จครับ และเก็บผลลัพธ์ หรือ Error ที่เกิดขึ้น
        for future in futures:
            try:
                future.result() 
            except Exception as e:
                print(f\"มีปัญหาตอนอัปโหลดไฟล์: {e} นะครับ\")

# ตัวอย่างการใช้งานจริง
# สร้างไฟล์ dummy หลายๆ อันก่อนนะ เพื่อทดสอบการอัปโหลดพร้อมกัน
# for i in range(5):
#     with open(f\"file_{i}_con.txt\", \"w\") as f:
#         f.write(f\"Content for concurrent file {i}\")
# file_list = [f\"file_{i}_con.txt\" for i in range(5)]
# upload_multiple_files(\"your-bucket-name\", file_list, \"my_multi_uploads_con/\")

5. เรื่องสำคัญเรื่องความปลอดภัย (Security)

เพื่อนๆ ต้องแน่ใจนะครับว่าได้ตั้งค่า Google Cloud credential อย่างปลอดภัย

เช่น ใช้ environment variable GOOGLE_APPLICATION_CREDENTIALS ชี้ไปที่ Service Account Key file ของเรา

หรือใช้ gcloud auth application-default login สำหรับการพัฒนาครับ

ห้าม Hardcode key หรือ credentials ลงในโค้ดโดยเด็ดขาดเลยนะครับ มันอันตรายมาก

สรุปนะครับ

ด้วยเทคนิคพวกนี้ เพื่อนๆ ก็สามารถสร้าง Python CLI ที่จัดการไฟล์บน GCS ได้อย่างมั่นใจมากขึ้นนะครับ ทั้งทนทานต่อปัญหา และยังอัปโหลดได้เร็วขึ้นด้วยครับ

ลองเอาไปปรับใช้กับงานของเพื่อนๆ ดูนะครับ ผมว่ามีประโยชน์แน่นอนครับ

อ้างอิง:

• Google Cloud Storage Client Library for Python
• Tenacity Library
• Concurrent.futures - High-level interface for asynchronously executing callables

หวังว่าบทความนี้จะเป็นประโยชน์กับทุกคนนะครับ

เขียนโดย cii3.net