pulumi-python

Pulumi Python 技能

开发工作流

1. 项目设置

# Create new Python project
pulumi new python

# Or with a cloud-specific template
pulumi new aws-python
pulumi new azure-python
pulumi new gcp-python

项目结构：

my-project/
├── Pulumi.yaml
├── __main__.py
├── requirements.txt     # or pyproject.toml
└── venv/                # Virtual environment

2. Pulumi ESC 集成

使用 Pulumi ESC 集中管理密钥和配置，而不是使用堆栈配置文件。

将 ESC 环境关联到堆栈：

# Create ESC environment
pulumi env init myorg/myproject-dev

# Edit environment
pulumi env edit myorg/myproject-dev

# Link to Pulumi stack
pulumi config env add myorg/myproject-dev

ESC 环境定义（YAML）：

values:
  # Static configuration
  pulumiConfig:
    aws:region: us-west-2
    myapp:instanceType: t3.medium

  # Dynamic OIDC credentials for AWS
  aws:
    login:
      fn::open::aws-login:
        oidc:
          roleArn: arn:aws:iam::123456789:role/pulumi-oidc
          sessionName: pulumi-deploy

  # Pull secrets from AWS Secrets Manager
  secrets:
    fn::open::aws-secrets:
      region: us-west-2
      login: ${aws.login}
      get:
        dbPassword:
          secretId: prod/database/password

  # Expose to environment variables
  environmentVariables:
    AWS_ACCESS_KEY_ID: ${aws.login.accessKeyId}
    AWS_SECRET_ACCESS_KEY: ${aws.login.secretAccessKey}
    AWS_SESSION_TOKEN: ${aws.login.sessionToken}

验证 ESC 环境：

# View resolved values (verify no errors)
pulumi env open myorg/myproject-dev

# Check for missing required values
pulumi env open myorg/myproject-dev --format json | jq .

3. Python 模式

创建基本资源：

import pulumi
import pulumi_aws as aws

# Get configuration from ESC
config = pulumi.Config()
instance_type = config.require("instanceType")

# Create resources with proper tagging
bucket = aws.s3.Bucket(
    "my-bucket",
    versioning=aws.s3.BucketVersioningArgs(
        enabled=True,
    ),
    server_side_encryption_configuration=aws.s3.BucketServerSideEncryptionConfigurationArgs(
        rule=aws.s3.BucketServerSideEncryptionConfigurationRuleArgs(
            apply_server_side_encryption_by_default=aws.s3.BucketServerSideEncryptionConfigurationRuleApplyServerSideEncryptionByDefaultArgs(
                sse_algorithm="AES256",
            ),
        ),
    ),
    tags={
        "Environment": pulumi.get_stack(),
        "ManagedBy": "Pulumi",
    },
)

# Export outputs
pulumi.export("bucket_name", bucket.id)
pulumi.export("bucket_arn", bucket.arn)

使用字典字面量（简洁的替代方式）：

import pulumi
import pulumi_aws as aws

bucket = aws.s3.Bucket(
    "my-bucket",
    versioning={"enabled": True},
    server_side_encryption_configuration={
        "rule": {
            "apply_server_side_encryption_by_default": {
                "sse_algorithm": "AES256",
            },
        },
    },
    tags={
        "Environment": pulumi.get_stack(),
        "ManagedBy": "Pulumi",
    },
)

使用组件资源实现复用：

import pulumi
from pulumi_aws import lb


class WebServiceArgs:
    def __init__(self, port: pulumi.Input[int], image_uri: pulumi.Input[str]):
        self.port = port
        self.image_uri = image_uri


class WebService(pulumi.ComponentResource):
    url: pulumi.Output[str]

    def __init__(self, name: str, args: WebServiceArgs, opts: pulumi.ResourceOptions = None):
        super().__init__("custom:app:WebService", name, {}, opts)

        # Create child resources with parent=self
        load_balancer = lb.LoadBalancer(
            f"{name}-lb",
            load_balancer_type="application",
            # ... configuration
            opts=pulumi.ResourceOptions(parent=self),
        )

        self.url = load_balancer.dns_name

        self.register_outputs({
            "url": self.url,
        })

使用堆栈引用处理跨堆栈依赖：

import pulumi

# Reference outputs from networking stack
networking_stack = pulumi.StackReference("myorg/networking/prod")
vpc_id = networking_stack.get_output("vpc_id")
subnet_ids = networking_stack.get_output("private_subnet_ids")

处理 Outputs：

import pulumi

# Apply transformation
uppercase_name = bucket.id.apply(lambda id: id.upper())

# Combine multiple outputs
combined = pulumi.Output.all(bucket.id, bucket.arn).apply(
    lambda args: f"Bucket {args[0]} has ARN {args[1]}"
)

# Using Output.concat for string interpolation
message = pulumi.Output.concat("Bucket ARN: ", bucket.arn)

# Conditional resources
is_prod = pulumi.get_stack() == "prod"
if is_prod:
    alarm = aws.cloudwatch.MetricAlarm(
        "alarm",
        # ... configuration
    )

4. 使用 pulumi env run 配合 ESC

运行任意命令，并注入 ESC 环境变量：

# Run pulumi commands with ESC credentials
pulumi env run myorg/aws-dev -- pulumi up

# Run tests with secrets
pulumi env run myorg/test-env -- pytest

# Open environment and export to shell
pulumi env open myorg/myproject-dev --format shell

5. 多语言组件

使用 Python 创建组件，以便在任何 Pulumi 语言（TypeScript、Go、C#、Java、YAML）中使用。

多语言组件的项目结构：

my-component/
├── PulumiPlugin.yaml      # Required for multi-language
├── pyproject.toml         # or requirements.txt
└── __main__.py            # Component + entry point

PulumiPlugin.yaml：

runtime: python

包含正确 Args 类的组件（main.py）：

from typing import Optional
import pulumi
from pulumi import Input, Output, ResourceOptions
from pulumi.provider.experimental import component_provider_host
import pulumi_aws as aws


class SecureBucketArgs:
    """Args class - use Input types for all properties."""

    def __init__(
        self,
        bucket_name: Input[str],
        enable_versioning: Optional[Input[bool]] = None,
        tags: Optional[Input[dict]] = None,
    ):
        self.bucket_name = bucket_name
        self.enable_versioning = enable_versioning or True
        self.tags = tags


class SecureBucket(pulumi.ComponentResource):
    """A secure S3 bucket with encryption and versioning."""

    bucket_id: Output[str]
    bucket_arn: Output[str]

    # Constructor must have 'args' parameter with type annotation
    def __init__(
        self,
        name: str,
        args: SecureBucketArgs,
        opts: Optional[ResourceOptions] = None,
    ):
        super().__init__("myorg:storage:SecureBucket", name, {}, opts)

        bucket = aws.s3.Bucket(
            f"{name}-bucket",
            bucket=args.bucket_name,
            versioning={"enabled": args.enable_versioning},
            server_side_encryption_configuration={
                "rule": {
                    "apply_server_side_encryption_by_default": {
                        "sse_algorithm": "AES256",
                    },
                },
            },
            tags=args.tags,
            opts=ResourceOptions(parent=self),
        )

        self.bucket_id = bucket.id
        self.bucket_arn = bucket.arn

        self.register_outputs({
            "bucket_id": self.bucket_id,
            "bucket_arn": self.bucket_arn,
        })


# Entry point for multi-language support
if __name__ == "__main__":
    component_provider_host(
        name="python-components",
        components=[SecureBucket],
    )

发布以供多语言使用：

# Consume from git repository
pulumi package add github.com/myorg/my-component

# With version tag
pulumi package add github.com/myorg/[email protected]

# Local development
pulumi package add /path/to/local/my-component

多语言 Args 要求：

• 对所有属性使用 pulumi.Input[T] 类型提示
• Args 类必须包含带类型化参数的 __init__
• 构造函数必须包含带类型注解的 args 参数
• 对可选属性使用 Optional[Input[T]]

**重要：**使用 component_provider_host()（来自 pulumi.provider.experimental）作为多语言组件的入口点。这是现代 API——较旧的 pulumi.provider.main() 搭配自定义 Provider 子类的方式已不再推荐用于组件开发。

6. 部署工作流

验证并部署，同时处理错误恢复：

# Step 1: Preview changes
pulumi preview

# Step 2: Review output for errors or unexpected changes
# If errors appear, fix code and re-run preview

# Step 3: Deploy only after preview succeeds
pulumi up

# Step 4: Verify outputs
pulumi stack output

用于基于 ESC 的部署：

# Step 1: Validate ESC environment resolves correctly
pulumi env open myorg/myproject-dev

# Step 2: Preview with ESC environment
pulumi env run myorg/myproject-dev -- pulumi preview

# Step 3: Deploy with ESC environment
pulumi env run myorg/myproject-dev -- pulumi up

# Step 4: Check stack outputs
pulumi stack output

配置与安全

• 所有密钥都使用 Pulumi ESC——切勿将密钥提交到堆栈配置文件或 Pulumi.yaml
• 启用 OIDC 身份验证，而不是使用静态凭据
• 尽可能使用 TTL 较短的动态密钥
• 为 OIDC 角色应用最小权限 IAM 策略
• 使用 ComponentResources 实现可复用的基础设施模式
• 利用 Python 类型提示，提升 IDE 支持并检测错误

常用命令

# Environment Commands (pulumi env)
pulumi env init <org>/<project>/<env>        # Create environment
pulumi env edit <org>/<env>                  # Edit environment
pulumi env open <org>/<env>                  # View resolved values
pulumi env run <org>/<env> -- <command>      # Run with env vars
pulumi env version tag <org>/<env> <tag>     # Tag version

# Pulumi Commands
pulumi new python                      # New project
pulumi config env add <org>/<env>     # Link ESC environment
pulumi preview                         # Preview changes
pulumi up                              # Deploy
pulumi stack output                    # View outputs
pulumi destroy                         # Tear down

# Dependency Management
pip install -r requirements.txt        # Install deps (pip)
poetry add pulumi-aws                  # Add dep (poetry)
uv add pulumi-aws                      # Add dep (uv)

Python 专属配置

使用 pip（默认）：

# Pulumi.yaml
runtime:
  name: python
  options:
    toolchain: pip
    virtualenv: venv

使用 poetry：

# Pulumi.yaml
runtime:
  name: python
  options:
    toolchain: poetry

使用 uv：

# Pulumi.yaml
runtime:
  name: python
  options:
    toolchain: uv
    virtualenv: .venv

启用类型检查：

# Pulumi.yaml
runtime:
  name: python
  options:
    typechecker: mypy  # or pyright

参考资料

• references/pulumi-esc.md - ESC 模式和命令
• references/pulumi-patterns.md - 常见基础设施模式
• references/pulumi-python.md - Python 专属指南
• references/pulumi-best-practices-aws.md - AWS 最佳实践
• references/pulumi-best-practices-azure.md - Azure 最佳实践
• references/pulumi-best-practices-gcp.md - GCP 最佳实践