本文目录导读：
我来介绍几种用Python生成Ansible Playbook的方法：
使用YAML库直接生成（推荐）
import yaml
def generate_playbook(hosts, tasks, name="Deploy Application"):
playbook = {
'name': name,
'hosts': hosts,
'become': True,
'vars': {
'app_name': 'myapp',
'app_port': 8080
},
'tasks': tasks
}
return playbook
# 生成包含多个play的playbook
def generate_multi_play_playbook():
playbooks = []
# Play 1: 安装nginx
play1 = {
'name': 'Install and configure nginx',
'hosts': 'webservers',
'become': True,
'tasks': [
{
'name': 'Install nginx',
'apt': {
'name': 'nginx',
'state': 'present',
'update_cache': True
}
},
{
'name': 'Start nginx',
'service': {
'name': 'nginx',
'state': 'started',
'enabled': True
}
}
]
}
playbooks.append(play1)
# Play 2: 部署应用
play2 = {
'name': 'Deploy application',
'hosts': 'appservers',
'become': True,
'tasks': [
{
'name': 'Create app directory',
'file': {
'path': '/opt/myapp',
'state': 'directory',
'mode': '0755'
}
},
{
'name': 'Copy app config',
'template': {
'src': 'app.conf.j2',
'dest': '/opt/myapp/app.conf'
}
}
]
}
playbooks.append(play2)
return playbooks
# 保存为YAML文件
def save_playbook(playbook_data, filename='deploy.yml'):
# 处理Ansible的特殊标签
class AnsibleDumper(yaml.Dumper):
pass
def represent_none(self, _):
return self.represent_scalar('tag:yaml.org,2002:null', 'None')
AnsibleDumper.add_representer(type(None), represent_none)
with open(filename, 'w', encoding='utf-8') as f:
if isinstance(playbook_data, list):
yaml.dump(playbook_data, f, Dumper=AnsibleDumper, default_flow_style=False, allow_unicode=True)
else:
yaml.dump([playbook_data], f, Dumper=AnsibleDumper, default_flow_style=False, allow_unicode=True)
# 使用示例
if __name__ == "__main__":
tasks = [
{
'name': 'Ensure package is installed',
'apt': {
'name': 'git',
'state': 'present'
}
},
{
'name': 'Clone repository',
'git': {
'repo': 'https://github.com/example/repo.git',
'dest': '/opt/app'
}
}
]
playbook = generate_playbook('webservers', tasks)
save_playbook(playbook)
使用Jinja2模板生成
from jinja2 import Environment, FileSystemLoader
import os
# 模板文件: playbook_template.j2
jinja_template = """
---
- name: {{ play_name }}
hosts: {{ hosts }}
become: {{ become | default(true) }}
vars:
app_name: "{{ app_name }}"
app_port: {{ app_port }}
tasks:
{% for task in tasks %}
- name: {{ task.name }}
{{ task.module }}:
{% for key, value in task.args.items() %}
{{ key }}: "{{ value }}"
{% endfor %}
{% if task.notify is defined %}
notify: {{ task.notify }}
{% endif %}
{% endfor %}
handlers:
{% for handler in handlers %}
- name: {{ handler.name }}
{{ handler.module }}:
{% for key, value in handler.args.items() %}
{{ key }}: "{{ value }}"
{% endfor %}
{% endfor %}
"""
def generate_playbook_with_jinja2(playbook_data, template_str):
env = Environment()
template = env.from_string(template_str)
return template.render(**playbook_data)
# 使用示例
handlers = [
{
'name': 'Restart nginx',
'module': 'service',
'args': {
'name': 'nginx',
'state': 'restarted'
}
}
]
tasks = [
{
'name': 'Install nginx',
'module': 'apt',
'args': {
'name': 'nginx',
'state': 'present'
},
'notify': 'Restart nginx'
}
]
playbook_content = generate_playbook_with_jinja2({
'play_name': 'WebServer Setup',
'hosts': 'webservers',
'become': True,
'app_name': 'myapp',
'app_port': 8080,
'tasks': tasks,
'handlers': handlers
})
# 保存文件
with open('playbook.yml', 'w') as f:
f.write(playbook_content)
动态生成复杂Playbook
import yaml
from datetime import datetime
class PlaybookGenerator:
def __init__(self):
self.plays = []
def add_play(self, name, hosts, become=True, vars=None, tasks=None, handlers=None):
play = {
'name': name,
'hosts': hosts,
'become': become
}
if vars:
play['vars'] = vars
if tasks:
play['tasks'] = tasks
if handlers:
play['handlers'] = handlers
self.plays.append(play)
def add_task(self, play_index, name, module, args, notify=None):
task = {
'name': name,
module: args
}
if notify:
task['notify'] = notify
if play_index < len(self.plays):
if 'tasks' not in self.plays[play_index]:
self.plays[play_index]['tasks'] = []
self.plays[play_index]['tasks'].append(task)
def add_role(self, play_index, role_name):
if play_index < len(self.plays):
if 'roles' not in self.plays[play_index]:
self.plays[play_index]['roles'] = []
self.plays[play_index]['roles'].append(role_name)
def save(self, filename=None):
if not filename:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"playbook_{timestamp}.yml"
with open(filename, 'w') as f:
yaml.dump(self.plays, f, default_flow_style=False, allow_unicode=True)
return filename
# 使用示例
generator = PlaybookGenerator()
# Play 1: Web服务器配置
generator.add_play(
name="Configure Web Servers",
hosts="webservers",
vars={
'http_port': 80,
'max_clients': 200
}
)
generator.add_task(
0,
name="Install nginx",
module="apt",
args={'name': 'nginx', 'state': 'present'},
notify="Restart nginx"
)
generator.add_task(
0,
name="Configure nginx",
module="template",
args={
'src': 'nginx.conf.j2',
'dest': '/etc/nginx/nginx.conf'
},
notify="Restart nginx"
)
generator.add_role(0, "security")
# Play 2: 数据库配置
generator.add_play(
name="Configure Database",
hosts="dbservers",
become=False
)
generator.add_task(
1,
name="Install PostgreSQL",
module="apt",
args={'name': 'postgresql', 'state': 'present'}
)
# 保存playbook
filename = generator.save()
print(f"Playbook saved as: {filename}")
从JSON/CSV导入生成
import yaml
import json
import csv
import os
def generate_playbook_from_json(json_file):
with open(json_file, 'r') as f:
data = json.load(f)
playbook = []
for play_data in data:
play = {
'name': play_data.get('name', 'Default Play'),
'hosts': play_data.get('hosts', 'all'),
'become': play_data.get('become', True),
'tasks': []
}
for task_data in play_data.get('tasks', []):
task = {
'name': task_data['name'],
task_data['module']: task_data['args']
}
play['tasks'].append(task)
playbook.append(play)
return playbook
def generate_playbook_from_csv(csv_file):
with open(csv_file, 'r') as f:
reader = csv.DictReader(f)
tasks = []
for row in reader:
task = {
'name': row['task_name'],
row['module']: {
'name': row['package_name'],
'state': row.get('state', 'present')
}
}
tasks.append(task)
playbook = [{
'name': 'Generated from CSV',
'hosts': 'all',
'become': True,
'tasks': tasks
}]
return playbook
# 使用示例
def generate_from_csv_example():
csv_data = """task_name,module,package_name,state
Install git,apt,git,present
Install curl,apt,curl,present
Install vim,apt,vim,present"""
with open('packages.csv', 'w') as f:
f.write(csv_data)
playbook = generate_playbook_from_csv('packages.csv')
with open('from_csv_playbook.yml', 'w') as f:
yaml.dump(playbook, f, default_flow_style=False)
os.remove('packages.csv')
这些方法各有特点：
- YAML库：最直接，适合静态定义
- Jinja2模板：灵活，适合动态生成
- PlaybookGenerator类：面向对象，适合复杂场景
- CSV/JSON导入：适合数据驱动的生成
选择哪种方法取决于你的具体需求和使用场景。
