Infrastructure as Code with Pulumi and Python: DanubeData Complete Guide

Python developers love Pulumi because it lets you define infrastructure using the same language you build applications with. No YAML, no HCL—just Python. This guide shows you how to manage your DanubeData infrastructure with clean, Pythonic code.

Why Pulumi with Python?

Python is a natural fit for infrastructure as code:

• Use familiar Python constructs: loops, conditionals, classes, dataclasses
• Leverage the entire Python ecosystem (requests, boto3, etc.)
• Type hints with mypy for safer infrastructure code
• Virtual environments for dependency isolation
• Integration with Jupyter notebooks for exploration

Getting Started

Install Pulumi CLI

# macOS
brew install pulumi

# Linux
curl -fsSL https://get.pulumi.com | sh

# Windows
choco install pulumi

Create a New Project

# Create a new directory
mkdir my-infrastructure && cd my-infrastructure

# Create and activate virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venvScriptsactivate

# Initialize a new Python project
pulumi new python

# Install the DanubeData provider
pip install pulumi_danubedata

Configure Authentication

# Set your DanubeData API token
pulumi config set danubedata:apiToken your-api-token --secret

Project Structure

A well-organized Pulumi Python project:

my-infrastructure/
├── Pulumi.yaml              # Project configuration
├── Pulumi.dev.yaml          # Development stack config
├── Pulumi.prod.yaml         # Production stack config
├── __main__.py              # Main entry point
├── requirements.txt         # Python dependencies
├── venv/                    # Virtual environment
└── infra/
    ├── __init__.py
    ├── vps.py               # VPS resources
    ├── database.py          # Database resources
    ├── cache.py             # Cache resources
    └── storage.py           # Storage resources

Creating Your First VPS

# __main__.py
import pulumi
import pulumi_danubedata as danubedata

# Create an SSH key for authentication
ssh_key = danubedata.SshKey("deploy-key",
    name="deploy-key",
    public_key="ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIExample... deploy@example.com"
)

# Create a VPS instance
web_server = danubedata.Vps("web-server",
    name="web-server",
    image="ubuntu-24.04",
    datacenter="fsn1",
    resource_profile="small_shared",  # 2 vCPU, 4GB RAM
    auth_method="ssh_key",
    ssh_key_id=ssh_key.id,
    enable_ipv6=True,
    enable_backups=True
)

# Export the VPS public IP
pulumi.export("web_server_ip", web_server.public_ip)
pulumi.export("web_server_ipv6", web_server.ipv6_address)

Complete Web Application Stack

Here's a complete example using Python best practices:

# __main__.py
import pulumi
from pulumi import Config, Output, export, get_stack
import pulumi_danubedata as danubedata
from dataclasses import dataclass
from typing import Optional

# Configuration
config = Config()
environment = get_stack()
app_name = config.require("app_name")

@dataclass
class ResourceProfiles:
    """Resource profiles for different environments."""
    vps: str
    database: str
    cache: str

    @classmethod
    def for_environment(cls, env: str) -> "ResourceProfiles":
        profiles = {
            "dev": cls("nano_shared", "small", "micro"),
            "staging": cls("small_shared", "small", "small"),
            "prod": cls("medium_dedicated", "medium", "medium"),
        }
        return profiles.get(env, profiles["dev"])

profiles = ResourceProfiles.for_environment(environment)

# =====================
# SSH Key
# =====================
ssh_key = danubedata.SshKey("deploy-key",
    name=f"{app_name}-{environment}-deploy",
    public_key=config.require_secret("ssh_public_key")
)

# =====================
# PostgreSQL Database
# =====================
database = danubedata.Database("app-db",
    name=f"{app_name}-{environment}-db",
    engine="postgresql",
    engine_version="16",
    resource_profile=profiles.database,
    database_name=app_name.replace("-", "_"),
    datacenter="fsn1",
    enable_ssl=True
)

# =====================
# Redis Cache
# =====================
cache = danubedata.Cache("app-cache",
    name=f"{app_name}-{environment}-cache",
    cache_provider="redis",
    resource_profile=profiles.cache,
    datacenter="fsn1"
)

# =====================
# Storage Bucket
# =====================
bucket = danubedata.StorageBucket("app-storage",
    name=f"{app_name}-{environment}-assets",
    region="fsn1",
    versioning_enabled=True
)

storage_keys = danubedata.StorageAccessKey("app-storage-key",
    name=f"{app_name}-{environment}-key"
)

# =====================
# Application VPS
# =====================
app_server = danubedata.Vps("app-server",
    name=f"{app_name}-{environment}-app",
    image="ubuntu-24.04",
    datacenter="fsn1",
    resource_profile=profiles.vps,
    auth_method="ssh_key",
    ssh_key_id=ssh_key.id,
    enable_ipv6=True,
    enable_backups=environment == "prod"
)

# =====================
# Firewall Rules
# =====================
admin_ip = config.require("admin_ip")

web_firewall = danubedata.Firewall("web-firewall",
    name=f"{app_name}-{environment}-web-fw",
    rules=[
        {
            "direction": "inbound",
            "protocol": "tcp",
            "ports": "80",
            "source_ips": ["0.0.0.0/0", "::/0"],
            "description": "Allow HTTP",
        },
        {
            "direction": "inbound",
            "protocol": "tcp",
            "ports": "443",
            "source_ips": ["0.0.0.0/0", "::/0"],
            "description": "Allow HTTPS",
        },
        {
            "direction": "inbound",
            "protocol": "tcp",
            "ports": "22",
            "source_ips": [admin_ip],
            "description": "SSH from admin only",
        },
    ]
)

# Database firewall - allow only from app server
db_firewall = danubedata.Firewall("db-firewall",
    name=f"{app_name}-{environment}-db-fw",
    rules=[
        {
            "direction": "inbound",
            "protocol": "tcp",
            "ports": "5432",
            "source_ips": [app_server.public_ip.apply(lambda ip: f"{ip}/32")],
            "description": "PostgreSQL from app server",
        },
    ]
)

# =====================
# Exports
# =====================
export("app_server_ip", app_server.public_ip)
export("app_server_ipv6", app_server.ipv6_address)
export("database_endpoint", database.endpoint)
export("database_name", database.database_name)
export("cache_endpoint", cache.endpoint)
export("storage_endpoint", bucket.endpoint_url)
export("storage_access_key", storage_keys.access_key)
export("storage_secret_key", storage_keys.secret_key)

Using Component Resources

Create reusable Python classes for complex infrastructure:

# infra/webapp.py
import pulumi
from pulumi import ComponentResource, ResourceOptions, Output
import pulumi_danubedata as danubedata
from dataclasses import dataclass
from typing import Optional


@dataclass
class WebAppArgs:
    """Arguments for creating a WebApp component."""
    name: str
    environment: str
    ssh_public_key: pulumi.Input[str]
    vps_profile: str = "small_shared"
    db_profile: str = "small"
    cache_profile: str = "micro"
    enable_backups: bool = False


class WebApp(ComponentResource):
    """A complete web application infrastructure component."""

    def __init__(self, name: str, args: WebAppArgs,
                 opts: Optional[ResourceOptions] = None):
        super().__init__("custom:resource:WebApp", name, {}, opts)

        child_opts = ResourceOptions(parent=self)

        # SSH Key
        self.ssh_key = danubedata.SshKey(f"{name}-ssh",
            name=f"{args.name}-{args.environment}",
            public_key=args.ssh_public_key,
            opts=child_opts
        )

        # Database
        self.database = danubedata.Database(f"{name}-db",
            name=f"{args.name}-{args.environment}-db",
            engine="postgresql",
            engine_version="16",
            resource_profile=args.db_profile,
            database_name=args.name.replace("-", "_"),
            datacenter="fsn1",
            opts=child_opts
        )

        # Cache
        self.cache = danubedata.Cache(f"{name}-cache",
            name=f"{args.name}-{args.environment}-cache",
            cache_provider="redis",
            resource_profile=args.cache_profile,
            datacenter="fsn1",
            opts=child_opts
        )

        # Storage
        self.bucket = danubedata.StorageBucket(f"{name}-storage",
            name=f"{args.name}-{args.environment}-assets",
            region="fsn1",
            opts=child_opts
        )

        # VPS
        self.vps = danubedata.Vps(f"{name}-vps",
            name=f"{args.name}-{args.environment}",
            image="ubuntu-24.04",
            datacenter="fsn1",
            resource_profile=args.vps_profile,
            auth_method="ssh_key",
            ssh_key_id=self.ssh_key.id,
            enable_backups=args.enable_backups,
            opts=child_opts
        )

        self.register_outputs({
            "vps_ip": self.vps.public_ip,
            "db_endpoint": self.database.endpoint,
            "cache_endpoint": self.cache.endpoint,
        })

# __main__.py
from infra.webapp import WebApp, WebAppArgs

app = WebApp("myapp", WebAppArgs(
    name="myapp",
    environment="prod",
    ssh_public_key="ssh-ed25519 AAAA...",
    vps_profile="medium_shared",
    db_profile="medium",
    enable_backups=True
))

pulumi.export("app_ip", app.vps.public_ip)
pulumi.export("db_endpoint", app.database.endpoint)

Creating Multiple Resources with Loops

# __main__.py
import pulumi
import pulumi_danubedata as danubedata

# Define your servers
servers = [
    {"name": "web-1", "profile": "small_shared"},
    {"name": "web-2", "profile": "small_shared"},
    {"name": "api-1", "profile": "medium_shared"},
    {"name": "worker-1", "profile": "medium_dedicated"},
]

# Create SSH key once
ssh_key = danubedata.SshKey("shared-key",
    name="shared-deploy-key",
    public_key="ssh-ed25519 AAAA..."
)

# Create all servers
vps_instances = {}
for server in servers:
    vps = danubedata.Vps(server["name"],
        name=server["name"],
        image="ubuntu-24.04",
        datacenter="fsn1",
        resource_profile=server["profile"],
        auth_method="ssh_key",
        ssh_key_id=ssh_key.id
    )
    vps_instances[server["name"]] = vps

# Export all IPs
for name, vps in vps_instances.items():
    pulumi.export(f"{name}_ip", vps.public_ip)

Serverless Containers

# serverless.py
import pulumi
import pulumi_danubedata as danubedata

# Deploy from Docker image
api_service = danubedata.Serverless("api",
    name="api-service",
    deployment_type="image",
    image_url="ghcr.io/myorg/api:latest",
    port=8080,
    min_instances=0,  # Scale to zero
    max_instances=10,
    memory=512,
    cpu=0.5,
    env_vars={
        "NODE_ENV": "production",
        "DATABASE_URL": database.endpoint,
        "REDIS_URL": cache.endpoint,
    }
)

# Deploy from Git repository
web_app = danubedata.Serverless("web",
    name="web-app",
    deployment_type="git",
    git_url="https://github.com/myorg/web-app.git",
    git_branch="main",
    port=3000,
    min_instances=1,
    max_instances=5,
    build_command="npm run build",
    start_command="npm start"
)

pulumi.export("api_url", api_service.url)
pulumi.export("web_url", web_app.url)

Using Data Sources

# data_sources.py
import pulumi
import pulumi_danubedata as danubedata

# Get available VPS images
images = danubedata.get_vps_images()
for image in images.images:
    print(f"Image: {image.name} - {image.slug}")

# Get existing SSH keys
existing_keys = danubedata.get_ssh_keys()

# Get existing VPS instances
existing_vps = danubedata.get_vpss()
for vps in existing_vps.instances:
    pulumi.log.info(f"VPS: {vps.name} - {vps.public_ip}")

# Get all databases
databases = danubedata.get_databases()

Type Hints with Mypy

# infra/types.py
from typing import TypedDict, List, Optional, Literal
import pulumi

class FirewallRule(TypedDict):
    direction: Literal["inbound", "outbound"]
    protocol: Literal["tcp", "udp", "icmp"]
    ports: str
    source_ips: List[str]
    description: Optional[str]

class ServerConfig(TypedDict):
    name: str
    profile: str
    image: str
    enable_backups: bool

def create_firewall_rules(rules: List[FirewallRule]) -> List[dict]:
    """Create validated firewall rules."""
    return [dict(rule) for rule in rules]

Dynamic Providers

Create custom resources using Python:

# custom_resources.py
from typing import Any, Optional
import pulumi
from pulumi.dynamic import Resource, ResourceProvider, CreateResult, UpdateResult
import requests

class DnsRecordProvider(ResourceProvider):
    """Custom provider for external DNS records."""

    def create(self, props: dict) -> CreateResult:
        # Call external API
        response = requests.post(
            f"https://api.dns-provider.com/records",
            json={
                "name": props["name"],
                "type": props["type"],
                "value": props["value"],
                "ttl": props.get("ttl", 300),
            },
            headers={"Authorization": f"Bearer {props['api_key']}"}
        )
        record_id = response.json()["id"]
        return CreateResult(id_=record_id, outs=props)

    def delete(self, id: str, props: dict) -> None:
        requests.delete(
            f"https://api.dns-provider.com/records/{id}",
            headers={"Authorization": f"Bearer {props['api_key']}"}
        )

class DnsRecord(Resource):
    def __init__(self, name: str, props: dict,
                 opts: Optional[pulumi.ResourceOptions] = None):
        super().__init__(DnsRecordProvider(), name, props, opts)

CI/CD with GitHub Actions

# .github/workflows/pulumi.yml
name: Pulumi Deploy

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

env:
  PULUMI_ACCESS_TOKEN: ${{ secrets.PULUMI_ACCESS_TOKEN }}
  DANUBEDATA_API_TOKEN: ${{ secrets.DANUBEDATA_API_TOKEN }}

jobs:
  preview:
    if: github.event_name == 'pull_request'
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install -r requirements.txt
      - uses: pulumi/actions@v5
        with:
          command: preview
          stack-name: dev

  deploy:
    if: github.ref == 'refs/heads/main'
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install -r requirements.txt
      - uses: pulumi/actions@v5
        with:
          command: up
          stack-name: prod

Testing Infrastructure Code

# test_infrastructure.py
import unittest
import pulumi

class InfrastructureTest(unittest.TestCase):
    @pulumi.runtime.test
    def test_vps_has_correct_profile(self):
        from __main__ import app_server

        def check_profile(profile):
            self.assertEqual(profile, "small_shared")

        return app_server.resource_profile.apply(check_profile)

    @pulumi.runtime.test
    def test_database_has_ssl_enabled(self):
        from __main__ import database

        def check_ssl(ssl_enabled):
            self.assertTrue(ssl_enabled)

        return database.enable_ssl.apply(check_ssl)

if __name__ == "__main__":
    unittest.main()

Useful Commands

# Preview changes
pulumi preview

# Apply changes
pulumi up

# View outputs
pulumi stack output

# Export to JSON
pulumi stack output --json

# Destroy resources
pulumi destroy

# Import existing resource
pulumi import danubedata:index:Vps my-vps existing-vps-id

Requirements File

# requirements.txt
pulumi>=3.0.0
pulumi_danubedata>=1.0.0

Cost Estimation

Get Started

Ready to manage your infrastructure with Python?

Create your free DanubeData account

Then install the Pulumi provider:

pip install pulumi_danubedata

Full documentation and examples are available in our GitHub repository.