Fabricの概要と文字エンコーディング設定
PythonのFabricは、ローカルタスク実行を担うinvokeとSSH/SFTP処理を扱うparamikoを統合したインフラ自動化フレームワークです。Windows環境で日本語出力を含むコマンドを実行する場合は、デフォルトのエンコーディングを明示的にUTF-8へ切り替える必要があります。
import invoke.runners
invoke.runners.default_encoding = lambda: "utf-8"Invokeによるタスク定義とオプション制御
invokeモジュールは、コマンドライン引数の解析、タスクの登録、ローカルシェルの実行を提供します。デコレータを使用して柔軟なオプション定義が可能です。
CLIオプションのパターン
オプショナル引数、繰り返し指定可能な値、累積カウント、フラグ型、およびダッシュとアンダースコアの自動変換に対応しています。
from invoke import task, call
@task(optional=['debug_log'])
def setup_env(ctx, debug_log=None):
target_log = debug_log if isinstance(debug_log, str) else "/var/log/setup.log"
print(f"Logging to: {target_log}")
@task(iterable=['tag'])
def filter_items(ctx, tag):
print(f"Processing tag: {tag}")
@task(incrementable=['quiet_level'])
def process_data(ctx, quiet_level=0):
print(f"Verbosity suppressed by {quiet_level} levels")
@task
def toggle_color(ctx, color=True):
# --no-color で False に切り替え可能
print(f"Color output enabled: {color}")実行例:
$ inv setup_env --debug-log=custom.log
$ inv filter_items --tag=dev --tag=prod
$ inv process_data -qqq # quiet_level=3
$ inv toggle_color --no-colorタスクの依存関係とワークフロー
タスク実行前後に別のタスクをフックさせることで、処理パイプラインを構築できます。
@task
def cleanup(ctx):
ctx.run("rm -rf /tmp/build_cache")
@task
def notify(ctx):
ctx.run("echo Deployment complete | mail admin@example.com")
@task(pre=[cleanup], post=[notify])
def deploy(ctx):
ctx.run("echo Deploying application...")
@task(pre=[call(setup_env, debug_log="/tmp/debug.log")])
def init_system(ctx):
print("System initialized.")コンテキスト操作とテスト用モック
作業ディレクトリの移動や環境変数のプリフィックス設定はコンテキストマネージャで行えます。また、ユニットテスト用に実行結果を偽装するモック機能も提供されます。
# ディレクトリ移動とコマンド前処理のネスト
with ctx.cd("/opt/webapp"):
with ctx.prefix("source /opt/venv/bin/activate"):
ctx.run("python manage.py migrate")
ctx.run("gunicorn app.wsgi -b 0.0.0.0:8000")
# モックコンテキストの定義
from invoke import MockContext, Result
mock_ctx = MockContext(run={"check_status": Result(stdout="OK\n", exited=0)})
assert mock_ctx.run("check_status").stdout == "OK\n"
# 動的な結果登録
mock_ctx.set_result_for("run", "fetch_data", Result(stdout='{"status":200}\n'))Paramikoを活用した基盤通信
Fabricの高レベルAPIを使用しない場合、paramikoを直接呼び出してSSHコマンド実行やファイル転送を制御できます。
import paramiko
# SSHコマンド実行
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect("10.0.1.50", username="deploy", password="SecurePass1!")
_, stdout, stderr = client.exec_command("uname -r")
print(stdout.read().decode().strip())
client.close()
# SFTPによるファイル転送
transport = paramiko.Transport(("10.0.1.50", 22))
transport.connect(username="deploy", password="SecurePass1!")
sftp = paramiko.SFTPClient.from_transport(transport)
sftp.put("local_config.yaml", "/etc/app/config.yaml")
sftp.get("/var/log/service.log", "downloaded_service.log")
transport.close()Fabricコネクションの構成と認証
接続情報の管理はコード内パラメータまたは外部設定ファイルで一元化できます。SSH鍵認証とパスワード認証の両方をサポートします。
from fabric import Connection
# SSH鍵認証
conn_key = Connection("deploy@10.0.1.50", connect_kwargs={"key_filename": "~/.ssh/id_ed25519"})
# パスワード認証
conn_pass = Connection("deploy@10.0.1.50", connect_kwargs={"password": "SecurePass1"})プロジェクトルートまたはユーザーホームディレクトリにfabric.ymlを配置すると、接続パラメータを自動的に読み込みます。
connect_kwargs:
password: SecurePass1
user: deploy
tasks:
collection_name: fabtasks接続メソッドの活用パターン
Connectionオブジェクトはリモート/ローカルのコマンド実行、ファイル同期、ポートフォワーディングを提供します。
with Connection("app-server-01") as c:
c.run("systemctl status nginx")
c.put("dist/index.html", "/var/www/html/")
c.get("/var/log/app/current.log", "app.log")
# ローカルポートをリモートポートへ転送
with c.forward_local(8080, remote_port=3000):
c.local("curl -s http://127.0.0.1:8080/api/health")
# リモートポートをローカルポートへ転送
with c.forward_remote(5543, local_port=5432):
c.run("pg_dump -h localhost -p 5543 -U dbuser mydb > backup.sql")runメソッドの高度な制御パラメータ
コマンド実行の挙動は多数のキーワード引数で微調整可能です。
- 非同期実行 (
asynchronous): 即時リターンし、後から結果を結合できます。 - 出力制御 (
hide):"out","err","both",Trueで標準出力/エラー出力を抑制。 - エラー無視 (
warn): 終了コードが0以外でも例外を発生させず処理を継続。 - インタラクティブ応答 (
watchers): 特定のパターンにマッチした際、自動で文字列を挿入。 - ストリームリダイレクト (
out_stream,err_stream): 出力をファイルオブジェクトへ直接書き込み。 - PTY制御 (
pty): 仮想端末の割り当て有無。対話型コマンドやos.environの継承に影響。 - シェル指定 (
shell): 実行シェルの変更（デフォルト/bin/bash）。 - タイムアウト (
timeout): 秒単位での実行制限。 - エコー制御 (
echo,dry): コマンド表示やドライランの実施。 - エンコーディング (
encoding): 出力デコード時の文字コード指定。
# 非同期実行と結果取得
promise = c.run("sleep 2 && df -h", asynchronous=True)
result = promise.join()
# インタラクティブなプロンプトへの自動応答
from invoke import Responder
auto_confirm = Responder(
pattern=r"Proceed with removal\? \[y/N\]",
response="y\n"
)
c.run("rm -rf /tmp/old_builds", watchers=[auto_confirm])
# 出力をファイルへ直接書き込み
c.run("tail -f /var/log/syslog", out_stream=open("syslog_tail.txt", "w"), hide=True, timeout=30)実行結果オブジェクトの取り扱い
runメソッドはResultインスタンスを返します。終了コード、出力内容、成否フラグをプロパティで参照できます。
output = c.run("cat /etc/hostname")
print(f"Return Code: {output.return_code}")
print(f"Success: {output.ok}")
if output.failed:
print(output.stderr)
# 出力末尾の行を取得
last_lines = output.tail("stdout", count=5)ローカルコマンドと特権実行
リモートホストだけでなく、制御側のマシンでコマンドを実行したり、sudoを介した昇権操作も可能です。
# ローカル環境変数を上書きして実行
c.local("echo $APP_ENV", env={"APP_ENV": "production"}, replace_env=False)
# sudoによる昇権実行（パスワード自動応答可能）
c.sudo("apt-get update", user="root", password="RootPass!")複数ホストのバッチ実行
SerialGroupまたはThreadingGroupを使用することで、複数のサーバーに対して同一コマンドを直列または並列に実行できます。
from fabric import SerialGroup, ThreadingGroup, GroupException
# 直列実行
servers = SerialGroup("web01", "web02", "web03", connect_kwargs={"password": "Pw!"})
servers.run("uptime")
# 並列実行（接続オブジェクトのリストから構築）
node_list = [
Connection("db01", connect_kwargs={"password": "Pw!"}),
Connection("db02", connect_kwargs={"password": "Pw!"}),
]
try:
with ThreadingGroup.from_connections(node_list) as pool:
pool_res = pool.run("systemctl restart postgresql", warn=True)
except GroupException as exc:
pool_res = exc.result
for conn, res in pool_res.items():
if hasattr(res, 'stdout'):
print(f"{conn.host}: {res.stdout.strip()}")CLIからのタスク起動
タスク定義ファイル（デフォルトfabfile.py）を作成後、fabコマンドで対話的または一括実行できます。
# 登録済みタスクの一覧表示
$ fab --list
# 特定ホストでのタスク実行（設定ファイルから認証情報を参照）
$ fab -H 10.0.1.50 deploy
# ユーザー指定とパスワードの入力プロンプト表示
$ fab -H admin@10.0.1.50 restart --prompt-for-login-password