本文目录导读：
我来详细介绍如何使用Python脚本生成Prefect配置。
基础配置生成
使用Python字典创建配置
from prefect import flow, task
from prefect.client.schemas.schedules import CronSchedule
from prefect.deployments import Deployment
import json
import yaml
from pathlib import Path
# 创建基础配置
def create_prefect_config():
"""生成Prefect基础配置"""
config = {
"profiles": {
"default": {
"settings": {
"PREFECT_API_URL": "http://localhost:4200/api",
"PREFECT_LOGGING_LEVEL": "INFO",
"PREFECT_AGENT_QUERY_INTERVAL": 5.0
}
},
"production": {
"settings": {
"PREFECT_API_URL": "https://api.prefect.cloud/api",
"PREFECT_LOGGING_LEVEL": "WARNING",
"PREFECT_AGENT_QUERY_INTERVAL": 10.0,
"PREFECT_API_KEY": "${PREFECT_API_KEY}"
}
}
},
"logging": {
"level": "INFO",
"format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s",
"handlers": {
"console": {
"class": "rich.logging.RichHandler"
},
"file": {
"class": "logging.handlers.RotatingFileHandler",
"filename": "prefect.log",
"maxBytes": 10485760,
"backupCount": 5
}
}
}
}
return config
# 保存为JSON配置文件
def save_config_as_json(config, filepath="prefect.json"):
with open(filepath, 'w') as f:
json.dump(config, f, indent=2)
print(f"Configuration saved to {filepath}")
# 保存为YAML配置文件
def save_config_as_yaml(config, filepath="prefect.yaml"):
with open(filepath, 'w') as f:
yaml.dump(config, f, default_flow_style=False)
print(f"Configuration saved to {filepath}")
动态生成部署配置
from prefect.deployments import Deployment
from prefect.filesystems import GitHub, S3, LocalFileSystem
from prefect.infrastructure import DockerContainer, Process, KubernetesJob
def create_deployment_config(flow_name, flow_path):
"""生成部署配置"""
# 配置存储
storage_config = {
"github": {
"repository": "https://github.com/your/repo",
"reference": "main",
"access_token": "${GITHUB_TOKEN}"
},
"s3": {
"bucket": "my-prefect-bucket",
"aws_access_key_id": "${AWS_ACCESS_KEY_ID}",
"aws_secret_access_key": "${AWS_SECRET_ACCESS_KEY}"
},
"local": {
"basepath": "/path/to/flows"
}
}
# 配置基础设施
infrastructure_config = {
"docker": {
"image": "python:3.9",
"env": {
"PREFECT_API_URL": "http://prefect-server:4200/api"
},
"volumes": ["/local/path:/container/path"],
"labels": {"app": "prefect-worker"}
},
"kubernetes": {
"customizations": {
"resources": {
"requests": {"cpu": "500m", "memory": "512Mi"},
"limits": {"cpu": "1000m", "memory": "1Gi"}
},
"env_from": [{"config_map_ref": {"name": "prefect-config"}}]
}
},
"process": {
"working_dir": "/opt/prefect/flows",
"stream_output": True
}
}
return {
"flow_name": flow_name,
"flow_path": flow_path,
"parameters": {},
"storage": storage_config,
"infrastructure": infrastructure_config,
"schedule": create_schedule_config()
}
def create_schedule_config():
"""生成调度配置"""
from datetime import datetime, timedelta
from prefect.client.schemas.schedules import (
CronSchedule,
IntervalSchedule,
RRuleSchedule
)
schedules = {
"daily_midnight": {
"type": "cron",
"cron": "0 0 * * *",
"timezone": "Asia/Shanghai"
},
"every_hour": {
"type": "interval",
"interval": timedelta(hours=1),
"anchor_date": datetime.now()
},
"weekdays_9am": {
"type": "rrule",
"rrule": "FREQ=WEEKLY;BYDAY=MO,TU,WE,TH,FR;BYHOUR=9;BYMINUTE=0"
}
}
return schedules
使用配置文件生成器
class PrefectConfigGenerator:
"""Prefect配置生成器"""
def __init__(self):
self.config = {
"profiles": {},
"logging": {},
"workers": [],
"deployments": []
}
def add_profile(self, name, settings):
"""添加配置文件"""
if "profiles" not in self.config:
self.config["profiles"] = {}
self.config["profiles"][name] = {"settings": settings}
return self
def add_worker(self, name, worker_type="process", pools=None, **kwargs):
"""添加工作者配置"""
worker_config = {
"name": name,
"type": worker_type,
"pools": pools or ["default"],
"settings": kwargs
}
self.config.setdefault("workers", []).append(worker_config)
return self
def add_deployment(self, flow_path, **kwargs):
"""添加部署配置"""
deployment = {
"flow_path": flow_path,
**kwargs
}
self.config.setdefault("deployments", []).append(deployment)
return self
def set_logging(self, **kwargs):
"""设置日志配置"""
self.config["logging"] = kwargs
return self
def add_environment_variables(self, env_vars):
"""添加环境变量"""
self.config.setdefault("environment", {}).update(env_vars)
return self
def generate_yaml(self, filepath="generated_prefect.yaml"):
"""生成YAML配置文件"""
import yaml
# 自定义YAML输出格式
class CustomDumper(yaml.SafeDumper):
pass
def str_representer(dumper, data):
if '\n' in data:
return dumper.represent_scalar('tag:yaml.org,2002:str', data, style='|')
return dumper.represent_scalar('tag:yaml.org,2002:str', data)
CustomDumper.add_representer(str, str_representer)
with open(filepath, 'w') as f:
yaml.dump(self.config, f, Dumper=CustomDumper, default_flow_style=False, sort_keys=False)
return filepath
def generate_json(self, filepath="generated_prefect.json"):
"""生成JSON配置文件"""
import json
with open(filepath, 'w') as f:
json.dump(self.config, f, indent=2)
return filepath
# 使用示例
def example_usage():
generator = PrefectConfigGenerator()
# 添加配置
config = (generator
.add_profile("production", {
"PREFECT_API_URL": "http://prefect-server:4200/api",
"PREFECT_LOGGING_LEVEL": "INFO"
})
.add_profile("development", {
"PREFECT_API_URL": "http://localhost:4200/api",
"PREFECT_LOGGING_LEVEL": "DEBUG"
})
.add_worker("worker-1", worker_type="process", pools=["pool-1"],
work_queues=["queue-1", "queue-2"],
prefetch_seconds=10)
.add_worker("worker-2", worker_type="docker",
image="prefecthq/prefect:2-python3.9",
networks=["prefect-network"])
.add_deployment(
flow_path="./flows/my_flow.py",
name="my-flow-deployment",
tags=["production", "critical"],
description="Main data processing flow",
schedule={
"cron": "0 */6 * * *",
"timezone": "UTC"
},
parameters={
"batch_size": 1000,
"max_retries": 3
}
)
.set_logging(level="INFO", format="%(asctime)s - %(name)s - %(levelname)s")
.add_environment_variables({
"DATABASE_URL": "${DATABASE_URL}",
"REDIS_URL": "${REDIS_URL}"
})
)
# 生成配置文件
yaml_file = config.generate_yaml("production_prefect.yaml")
json_file = config.generate_json("production_prefect.json")
return yaml_file, json_file
使用模板引擎生成配置
from jinja2 import Environment, FileSystemLoader
from typing import Dict, Any
import os
class TemplateConfigGenerator:
"""使用Jinja2模板生成配置"""
def __init__(self, template_dir="templates"):
self.env = Environment(
loader=FileSystemLoader(template_dir),
trim_blocks=True,
lstrip_blocks=True
)
def generate_from_template(self, template_name: str, context: Dict[str, Any]) -> str:
"""从模板生成配置"""
template = self.env.get_template(template_name)
return template.render(**context)
def generate_prefect_config(self, environment: str = "development") -> str:
"""生成Prefect配置"""
context = {
"environment": environment,
"api_url": self._get_api_url(environment),
"logging_level": "DEBUG" if environment == "development" else "WARNING",
"workers": self._get_workers_config(environment),
"deployments": self._get_deployments_config(environment),
"created_at": os.environ.get("BUILD_TIMESTAMP", "unknown")
}
return self.generate_from_template("prefect_config.yaml.j2", context)
def _get_api_url(self, environment: str) -> str:
urls = {
"development": "http://localhost:4200/api",
"staging": "https://staging-api.prefect.local/api",
"production": "https://api.prefect.cloud/api"
}
return urls.get(environment, urls["development"])
def _get_workers_config(self, environment: str) -> list:
base_config = {
"development": [
{"name": "dev-worker", "type": "process", "pools": ["dev-pool"]}
],
"staging": [
{"name": "staging-worker-1", "type": "process", "pools": ["staging-pool"]},
{"name": "staging-worker-2", "type": "docker", "pools": ["staging-pool"]}
],
"production": [
{"name": "prod-worker-1", "type": "kubernetes", "pools": ["prod-pool"]},
{"name": "prod-worker-2", "type": "kubernetes", "pools": ["prod-pool"]},
{"name": "prod-worker-3", "type": "kubernetes", "pools": ["prod-pool"]}
]
}
return base_config.get(environment, [])
def _get_deployments_config(self, environment: str) -> list:
"""获取部署配置"""
return [
{
"name": f"data-pipeline-{environment}",
"flow_path": "./flows/data_pipeline.py",
"schedule": "0 */6 * * *" if environment == "production" else None,
"tags": [environment, "data-pipeline"],
"parameters": {
"environment": environment,
"batch_size": 10000 if environment == "production" else 100
}
}
]
# 创建Jinja2模板
def create_prefect_template():
"""创建Prefect配置模板"""
template_content = """
# Prefect Configuration
# Environment: {{ environment }}
# Generated at: {{ created_at }}
profiles:
{{ environment }}:
settings:
PREFECT_API_URL: {{ api_url }}
PREFECT_LOGGING_LEVEL: {{ logging_level }}
PREFECT_AGENT_QUERY_INTERVAL: 10.0
logging:
level: {{ logging_level }}
format: "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
handlers:
console:
class: "rich.logging.RichHandler"
file:
class: "logging.handlers.RotatingFileHandler"
filename: "prefect_{{ environment }}.log"
maxBytes: 10485760
backupCount: 5
workers:
{% for worker in workers %}
- name: {{ worker.name }}
type: {{ worker.type }}
pools: {{ worker.pools }}
work_queues:
- "{{ environment }}-queue"
prefetch_seconds: 10
{% if worker.type == "docker" %}
docker:
image: "prefecthq/prefect:2-python3.9"
networks: ["prefect-{{ environment }}"]
{% endif %}
{% if worker.type == "kubernetes" %}
kubernetes:
namespace: "prefect-{{ environment }}"
image: "prefecthq/prefect:2-python3.9"
service_account_name: "prefect-worker"
resources:
requests:
cpu: "500m"
memory: "512Mi"
limits:
cpu: "1000m"
memory: "1Gi"
{% endif %}
{% endfor %}
deployments:
{% for deployment in deployments %}
- name: {{ deployment.name }}
flow_path: {{ deployment.flow_path }}
{% if deployment.schedule %}
schedule:
cron: {{ deployment.schedule }}
timezone: "UTC"
{% endif %}
tags: {{ deployment.tags }}
parameters:
{% for key, value in deployment.parameters.items() %}
{{ key }}: {{ value if not value is string else '"' + value + '"' }}
{% endfor %}
apply_to:
tags: {{ deployment.tags }}
{% endfor %}
environment:
PREFECT_SERVER_HOST: "0.0.0.0"
PREFECT_SERVER_PORT: 4200
PREFECT_UI_ENABLED: "true"
"""
# 保存模板
os.makedirs("templates", exist_ok=True)
with open("templates/prefect_config.yaml.j2", "w") as f:
f.write(template_content)
print("Template created: templates/prefect_config.yaml.j2")
完整的配置生成脚本
#!/usr/bin/env python3
"""Prefect配置生成器脚本"""
import argparse
import json
import yaml
import os
from pathlib import Path
from typing import Dict, Any
def create_complete_prefect_config():
"""创建完整的Prefect配置"""
return {
"profiles": {
"default": {
"settings": {
"PREFECT_API_URL": "http://localhost:4200/api",
"PREFECT_API_KEY": None
}
}
},
"logging": {
"level": "INFO",
"format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
},
"deployments": [
{
"name": "etl-pipeline",
"flow_path": "./flows/etl.py",
"parameters": {},
"schedule": {
"cron": "0 0 * * *",
"timezone": "UTC"
},
"tags": ["production", "etl"],
"storage": {
"type": "local",
"basepath": "/opt/prefect/flows"
},
"infrastructure": {
"type": "process",
"env": {
"PYTHONPATH": "${PYTHONPATH}:/opt/prefect"
},
"working_dir": "/opt/prefect"
}
}
],
"workers": [
{
"name": "worker-1",
"type": "process",
"pools": ["default"],
"work_queues": ["default"],
"prefetch_seconds": 10
}
],
"environment": {
"PREFECT_SERVER_HOST": "0.0.0.0",
"PREFECT_SERVER_PORT": 4200,
"PREFECT_UI_ENABLED": "true",
"PREFECT_API_DATABASE_CONNECTION_URL": "sqlite+aiosqlite:///./prefect.db",
"PREFECT_AGENT_PREFETCH_SECONDS": 5,
"PREFECT_LOGGING_TO_API_ENABLED": "true"
}
}
def save_config(config: Dict[str, Any], output_format: str = "yaml", output_path: str = None):
"""保存配置到文件"""
if output_path is None:
output_path = f"prefect_config.{output_format}"
if output_format == "yaml":
with open(output_path, 'w') as f:
yaml.dump(config, f, default_flow_style=False, sort_keys=False)
elif output_format == "json":
with open(output_path, 'w') as f:
json.dump(config, f, indent=2)
print(f"Configuration saved to {output_path}")
return output_path
def main():
parser = argparse.ArgumentParser(description="Generate Prefect configuration")
parser.add_argument("--format", choices=["yaml", "json"], default="yaml",
help="Output format (default: yaml)")
parser.add_argument("--output", "-o", help="Output file path")
parser.add_argument("--env", choices=["development", "staging", "production"],
default="development", help="Environment type")
args = parser.parse_args()
# 生成配置
config = create_complete_prefect_config()
# 根据环境调整配置
if args.env == "development":
config["logging"]["level"] = "DEBUG"
config["profiles"]["default"]["settings"]["PREFECT_API_URL"] = "http://localhost:4200/api"
elif args.env == "production":
config["logging"]["level"] = "WARNING"
config["profiles"]["production"] = {
"settings": {
"PREFECT_API_URL": "https://api.prefect.cloud/api",
"PREFECT_API_KEY": "${PREFECT_API_KEY}"
}
}
# 保存配置
save_config(config, args.format, args.output)
if __name__ == "__main__":
main()
使用示例
# 1. 基础配置生成
config = create_prefect_config()
save_config_as_yaml(config, "basic_prefect.yaml")
# 2. 使用配置生成器
generator = PrefectConfigGenerator()
config = (generator
.add_profile("production", {"PREFECT_API_URL": "http://localhost:4200/api"})
.add_worker("my-worker", type="process")
.generate_yaml("generated_config.yaml"))
# 3. 使用命令行
# python generate_prefect_config.py --format yaml --env production
# 4. 使用模板
generator = TemplateConfigGenerator()
config = generator.generate_prefect_config("production")
print(config)
这些方法可以帮助你灵活地生成Prefect配置文件,可以根据具体需求选择合适的方式。
