Giải thích chi tiết về mô-đun subprocess của Python｜Từ cơ bản đến nâng cao

By 佐川 直弘 | Naohiro Sagawa (運営者)

※記事内に広告を含みます。Amazonアソシエイトとして適格販売により収入を得ています。

1. Mô-đun subprocess của Python là gì?

Tổng quan

Mô-đun subprocess của Python là một công cụ mạnh mẽ giúp thực thi lệnh hệ thống hoặc chương trình bên ngoài từ Python. Bằng cách sử dụng mô-đun này, bạn có thể quản lý đầu vào/đầu ra chuẩn và kiểm soát tiến trình, giúp việc tích hợp giữa chương trình Python và các chương trình bên ngoài trở nên dễ dàng. Nó cung cấp phương pháp thay thế an toàn và linh hoạt hơn so với os.system() hay mô-đun commands.

Ứng dụng chính

• Thực thi lệnh shell: Gọi các lệnh hệ thống đơn giản.
• Quản lý tiến trình: Thực thi chương trình bên ngoài và chuyển hướng đầu vào/đầu ra chuẩn.
• Xử lý bất đồng bộ: Quản lý các tác vụ chạy trong thời gian dài hoặc chạy song song.

2. Cách sử dụng cơ bản: subprocess.run()

Cách sử dụng cơ bản

Hàm subprocess.run() cho phép thực thi lệnh hệ thống một cách đơn giản từ Python. Ví dụ, để liệt kê danh sách tệp trong thư mục, bạn có thể sử dụng đoạn mã sau:

import subprocess

result = subprocess.run(['ls', '-l'], capture_output=True, text=True)
print(result.stdout)

Đoạn mã trên thực thi lệnh ls -l và lưu đầu ra vào stdout, sau đó xử lý trong Python. Tham số capture_output=True giúp thu thập đầu ra chuẩn, và text=True cho phép xử lý kết quả dưới dạng chuỗi.

Xử lý lỗi

Khi sử dụng subprocess.run(), nếu lệnh gặp lỗi, bạn có thể sử dụng stderr để lấy thông báo lỗi. Đồng thời, có thể kiểm tra returncode để xác định lệnh có thực thi thành công hay không.

result = subprocess.run(['ls', 'nonexistentfile'], capture_output=True, text=True)
if result.returncode != 0:
    print(f"Lỗi: {result.stderr}")

Trong ví dụ này, nếu chỉ định một tệp không tồn tại, thông báo lỗi sẽ được hiển thị qua đầu ra lỗi chuẩn.

3. Thực thi bất đồng bộ với subprocess.Popen()

Xử lý bất đồng bộ với Popen

Hàm subprocess.run() thực thi lệnh theo kiểu đồng bộ, nghĩa là chương trình Python sẽ chờ lệnh kết thúc trước khi tiếp tục. Tuy nhiên, nếu bạn muốn thực thi tiến trình theo kiểu bất đồng bộ, có thể sử dụng subprocess.Popen(). Điều này giúp chương trình có thể thực hiện nhiều công việc cùng lúc.

import subprocess

proc = subprocess.Popen(['sleep', '5'], stdout=subprocess.PIPE)
print("Tiến trình đã bắt đầu")
proc.wait()
print("Tiến trình đã kết thúc")

Đoạn mã trên thực thi lệnh sleep 5 theo kiểu bất đồng bộ, cho phép các tác vụ khác tiếp tục thực hiện trong lúc lệnh đang chạy.

Kiểm soát đầu vào/đầu ra chuẩn

Với Popen, bạn có thể kiểm soát việc chuyển hướng đầu vào/đầu ra một cách chi tiết hơn. Ví dụ dưới đây đọc dữ liệu từ một tệp, xử lý bằng lệnh cat và ghi kết quả ra một tệp khác.

with open('input.txt', 'r') as infile, open('output.txt', 'w') as outfile:
    proc = subprocess.Popen(['cat'], stdin=infile, stdout=outfile)
    proc.wait()

Đoạn mã trên giúp chuyển hướng dữ liệu đầu vào từ một tệp, xử lý bằng cat và ghi kết quả vào tệp đầu ra.

4. Ví dụ thực tế: Viết script tự động hóa

Sao lưu tệp tự động

Mô-đun subprocess rất hữu ích trong quản trị hệ thống và tự động hóa các tác vụ định kỳ. Ví dụ dưới đây giúp sao lưu tệp vào một thư mục nhất định.

import subprocess

files_to_backup = ['file1.txt', 'file2.txt', 'file3.txt']
backup_dir = '/backup/directory/'

for file in files_to_backup:
    subprocess.run(['cp', file, backup_dir])

Đoạn mã trên sẽ sao chép danh sách các tệp vào thư mục sao lưu. Bằng cách viết các script như vậy, bạn có thể tự động hóa quy trình sao lưu một cách hiệu quả.

Sử dụng trong CI/CD Pipeline

Mô-đun subprocess cũng được sử dụng trong các hệ thống tích hợp liên tục (CI) và triển khai liên tục (CD). Ví dụ, nó có thể tự động thực thi các script kiểm thử và chỉ tiếp tục nếu kiểm thử thành công.

5. Bảo mật và thực hành tốt nhất

Rủi ro khi sử dụng shell=True

Tùy chọn shell=True cho phép thực thi lệnh thông qua shell, nhưng nó có thể gây ra rủi ro bảo mật, đặc biệt khi nhận dữ liệu đầu vào từ bên ngoài. Nếu không kiểm soát tốt, nó có thể dẫn đến các cuộc tấn công tiêm lệnh shell (Shell Injection). Để tránh rủi ro này, nên sử dụng shell=False khi có thể.

import subprocess

# Cách sử dụng an toàn (khuyến nghị)
subprocess.run(['ls', '-l'])

# Cách sử dụng nguy hiểm (cần cẩn trọng)
subprocess.run('ls -l', shell=True)

Hỗ trợ đa nền tảng

Các lệnh hệ thống có thể khác nhau tùy theo hệ điều hành. Do đó, bạn nên sử dụng mô-đun platform để phát hiện hệ điều hành và chọn lệnh phù hợp.

import platform
import subprocess

if platform.system() == "Windows":
    subprocess.run(['dir'], shell=True)
else:
    subprocess.run(['ls', '-l'])

6. Xử lý lỗi và gỡ lỗi

Lỗi phổ biến và cách khắc phục

Khi sử dụng subprocess, bạn có thể gặp lỗi như “Không tìm thấy tệp” hoặc “Không đủ quyền truy cập”. Để khắc phục, bạn có thể sử dụng stderr để thu thập thông tin lỗi và kiểm tra returncode để xác định nguyên nhân.

Mẹo gỡ lỗi

Thêm tham số check=True sẽ giúp phát hiện lỗi sớm bằng cách ném ra ngoại lệ nếu lệnh không thực thi thành công. Bạn cũng có thể ghi lại đầu ra và lỗi để phân tích khi cần.

import subprocess

try:
    result = subprocess.run(['ls', '-l'], check=True, capture_output=True, text=True)
    print(result.stdout)
except subprocess.CalledProcessError as e:
    print(f"Lỗi xảy ra: {e}")

7. Xử lý bất đồng bộ với asyncio

Xử lý bất đồng bộ với asyncio

Bằng cách kết hợp với asyncio, bạn có thể chạy nhiều tiến trình đồng thời và quản lý chúng một cách hiệu quả. Dưới đây là một ví dụ sử dụng asyncio để chạy lệnh ls theo cách bất đồng bộ.

import asyncio
import subprocess

async def run_command():
    proc = await asyncio.create_subprocess_exec('ls', '-l',
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE)

    stdout, stderr = await proc.communicate()

    if stdout:
        print(f'[stdout]\n{stdout.decode()}')
    if stderr:
        print(f'[stderr]\n{stderr.decode()}')

asyncio.run(run_command())

Đoạn mã trên giúp thực thi lệnh bất đồng bộ và thu thập kết quả từ đầu ra chuẩn và đầu ra lỗi chuẩn. Việc sử dụng asyncio giúp bạn quản lý tác vụ bất đồng bộ một cách hiệu quả hơn.