---

name: planetscale-sdk description: Integrates PlanetScale using mysql-connector-python with patterns for database branching, deploy requests, schema management, connection pooling, and insights for serverless MySQL. license: MIT compatibility: opencode metadata: version: "1.0.0" domain: coding triggers: planetscale, mysql connector, database branching, planetscale api, how do i connect to planetscale from python, serverless mysql, deploy request, vitess archetypes:

• tactical
• generation anti_triggers:
• brainstorming
• vague ideation
• code golf
• over-engineering response_profile: verbosity: low directive_strength: high abstraction_level: operational role: implementation scope: implementation output-format: code content-types:
• code
• guidance
• do-dont
• examples related-skills: coding-postgresql-sdk, coding-supabase-sdk, coding-database-migrations

---

PlanetScale Python SDK Integration

Integrates PlanetScale serverless MySQL using mysql-connector-python (and the PlanetScale API via HTTP) with patterns for database branching, deploy requests, schema management, connection with SSL, connection pooling for serverless environments, and querying with Python.

TL;DR Checklist

• Use mysql.connector.connect() with SSL for database connections — never without SSL
• Use mysql.connector.pooling.MySQLConnectionPool for connection pooling in web apps
• Use the PlanetScale HTTP API (/v1/databases, /v1/branches) for branch and deploy request management
• Use ALTER TABLE ... ONLINE for schema changes (PlanetScale's non-blocking DDL)
• Use GitHub integration for schema change deploy requests
• Use sqlparse or sqlglot for linting migrations before deploy requests
• Use SELECT /*+ SET_VAR(session_timeout=60) */ for long-running queries in production

---

When to Use

Use this skill when:

• Building Python applications that connect to PlanetScale serverless MySQL databases
• Implementing database branching workflows (development branches, production branches)
• Managing deploy requests to merge schema changes into production branches
• Writing connection code for serverless environments (AWS Lambda, Vercel, etc.)
• Performing database migrations with PlanetScale's non-blocking schema changes
• Querying MySQL-compatible databases through the PlanetScale proxy
• Integrating PlanetScale with SQLAlchemy or Django ORM

---

When NOT to Use

• For non-MySQL databases (use the appropriate SDK — PostgreSQL, MongoDB, etc.)
• For workloads that require FOREIGN KEY constraints (PlanetScale doesn't support them)
• For applications that need LOCK TABLES or similar MySQL features not supported by Vitess
• For high-volume OLTP that needs sub-millisecond latency (consider dedicated MySQL)
• When you need full-text search indexes (PlanetScale uses Vitess which has limits)

---

Core Workflow

1. Connect to PlanetScale with SSL

import mysql.connector
from mysql.connector import Error as MySQLError
from mysql.connector.pooling import MySQLConnectionPool
import os

# PlanetScale requires SSL — always use ssl_ca or ssl_mode
config = {
    "host": os.environ["PLANETSCALE_HOST"],        # e.g. us-east.connect.psdb.cloud
    "port": 3306,
    "database": os.environ["PLANETSCALE_DATABASE"],
    "user": os.environ["PLANETSCALE_USERNAME"],
    "password": os.environ["PLANETSCALE_PASSWORD"],
    "ssl_ca": "/etc/ssl/cert.pem",                 # Path to CA certificate
    "ssl_mode": "VERIFY_IDENTITY",                 # or "REQUIRED" if CA path fails
    "connection_timeout": 10,
    "pool_name": "planetscale_pool",
    "pool_size": 5,
}

# Single connection
conn = mysql.connector.connect(**config)

Checkpoint: Verify the connection with conn.is_connected(). PlanetScale requires SSL — never set ssl_mode to DISABLED. Use VERIFY_IDENTITY for production (prevents MITM) and REQUIRED if you can't configure the CA path.

2. Execute Queries

def query_orders(
    conn,
    customer_id: str,
    limit: int = 50,
) -> list[dict]:
    """Query orders with parameterized SQL."""
    query = """
        SELECT id, order_date, total, status, currency
        FROM orders
        WHERE customer_id = %s
        ORDER BY order_date DESC
        LIMIT %s
    """
    with conn.cursor(dictionary=True) as cursor:
        cursor.execute(query, (customer_id, limit))
        return cursor.fetchall()

def create_order(conn, order_data: dict) -> int:
    """Insert a new order and return its ID."""
    query = """
        INSERT INTO orders (customer_id, total, status, currency)
        VALUES (%(customer_id)s, %(total)s, %(status)s, %(currency)s)
    """
    with conn.cursor() as cursor:
        cursor.execute(query, order_data)
        conn.commit()
        return cursor.lastrowid

Checkpoint: PlanetScale runs MySQL 8.0+ via Vitess. Use cursor(dictionary=True) for dict results. Always call conn.commit() for write operations. PlanetScale autocommits DDL but explicit commit is needed for DML.

3. Connection Pooling for Serverless

class PlanetScalePool:
    """Thread-safe connection pool for PlanetScale."""

    def __init__(self, pool_size: int = 10):
        self._pool = MySQLConnectionPool(
            pool_name="pscale_pool",
            pool_size=pool_size,
            pool_reset_session=True,
            **get_connection_config(),
        )

    def execute(self, query: str, params: tuple | dict | None = None) -> list[dict]:
        """Acquire a connection, execute, return results."""
        conn = self._pool.get_connection()
        try:
            with conn.cursor(dictionary=True) as cursor:
                cursor.execute(query, params or ())
                if cursor.description:  # SELECT-like query
                    return cursor.fetchall()
                conn.commit()
                return []
        except MySQLError as exc:
            conn.rollback()
            raise RuntimeError(f"Query failed: {exc}") from exc
        finally:
            conn.close()  # Returns to pool

pool = PlanetScalePool()

def get_user_orders(user_id: str) -> list[dict]:
    return pool.execute(
        "SELECT * FROM orders WHERE customer_id = %s ORDER BY created_at DESC LIMIT 50",
        (user_id,),
    )

Checkpoint: PlanetScale's connection proxy handles serverless connection surges. Pool size of 5-10 is typically sufficient. Close connections with conn.close() to return them to the pool — never leave connections open in serverless environments.

4. PlanetScale API — Database Branches

import httpx

class PlanetScaleAPI:
    """HTTP client for PlanetScale management API."""

    BASE_URL = "https://api.planetscale.com/v1"

    def __init__(self, service_token: str):
        self._client = httpx.Client(
            base_url=self.BASE_URL,
            headers={
                "Authorization": f"Bearer {service_token}",
                "Content-Type": "application/json",
            },
        )

    def list_branches(self, org: str, db: str) -> list[dict]:
        """List all branches for a database."""
        resp = self._client.get(f"/organizations/{org}/databases/{db}/branches")
        resp.raise_for_status()
        return resp.json()["data"]

    def create_branch(self, org: str, db: str, name: str, parent: str = "main") -> dict:
        """Create a new database branch from a parent branch."""
        resp = self._client.post(
            f"/organizations/{org}/databases/{db}/branches",
            json={"name": name, "parent": parent},
        )
        resp.raise_for_status()
        return resp.json()["data"]

    def create_deploy_request(
        self,
        org: str,
        db: str,
        into_branch: str,
        from_branch: str,
        notes: str = "",
    ) -> dict:
        """Create a deploy request to merge schema changes."""
        resp = self._client.post(
            f"/organizations/{org}/databases/{db}/deploy-requests",
            json={
                "into_branch": into_branch,
                "from_branch": from_branch,
                "notes": notes,
            },
        )
        resp.raise_for_status()
        return resp.json()["data"]

    def deploy_request_status(self, org: str, db: str, dr_id: int) -> dict:
        """Check deploy request status (pending, ready, deploying, complete)."""
        resp = self._client.get(
            f"/organizations/{org}/databases/{db}/deploy-requests/{dr_id}",
        )
        resp.raise_for_status()
        return resp.json()["data"]

Checkpoint: The PlanetScale API requires a service token (not the database password). Create tokens in the PlanetScale dashboard under Settings > Service Tokens. Deploy requests have states: pending → ready → deploying → complete (or closed on conflict).

---

Implementation Patterns

Pattern 1: Migration Workflow with Deploy Requests

def run_migration(
    api: PlanetScaleAPI,
    org: str,
    db: str,
    branch_name: str,
    migration_sql: str,
) -> dict:
    """Run a schema migration through PlanetScale's branching workflow."""
    # 1. Create a branch for the migration
    branch = api.create_branch(org, db, branch_name)

    # 2. Connect to the branch and run migration
    branch_config = get_connection_config(branch_name=branch_name)
    conn = mysql.connector.connect(**branch_config)
    try:
        with conn.cursor() as cursor:
            for statement in migration_sql.split(";"):
                statement = statement.strip()
                if statement:
                    cursor.execute(statement)
    finally:
        conn.close()

    # 3. Create a deploy request
    dr = api.create_deploy_request(org, db, "main", branch_name, notes=f"Migration: {branch_name}")
    return {"branch": branch["name"], "deploy_request_id": dr["id"]}

Pattern 2: SQLAlchemy Integration

from sqlalchemy import create_engine, text

def create_planetscale_engine() -> Engine:
    """Create a SQLAlchemy engine for PlanetScale."""
    connection_string = (
        f"mysql+mysqlconnector://{os.environ['PLANETSCALE_USERNAME']}:"
        f"{os.environ['PLANETSCALE_PASSWORD']}@"
        f"{os.environ['PLANETSCALE_HOST']}:3306/"
        f"{os.environ['PLANETSCALE_DATABASE']}"
        "?ssl_ca=/etc/ssl/cert.pem&ssl_mode=VERIFY_IDENTITY"
    )
    return create_engine(connection_string, pool_size=5, max_overflow=10)

def query_with_sqlalchemy(engine: Engine, user_id: str) -> list[dict]:
    with engine.connect() as conn:
        result = conn.execute(
            text("SELECT * FROM orders WHERE customer_id = :uid ORDER BY created_at DESC"),
            {"uid": user_id},
        )
        return [row._asdict() for row in result]

BAD vs GOOD: Connection Pattern

# ❌ BAD — Connecting without SSL
def connect_bad():
    return mysql.connector.connect(
        host="us-east.connect.psdb.cloud",
        user="user",
        password="pass",
        database="mydb",
        # No SSL config — will fail on PlanetScale!
    )

# ✅ GOOD — SSL connection with VERIFY_IDENTITY
def connect_good():
    return mysql.connector.connect(
        host=os.environ["PLANETSCALE_HOST"],
        user=os.environ["PLANETSCALE_USERNAME"],
        password=os.environ["PLANETSCALE_PASSWORD"],
        database=os.environ["PLANETSCALE_DATABASE"],
        ssl_ca="/etc/ssl/cert.pem",
        ssl_mode="VERIFY_IDENTITY",
    )

BAD vs GOOD: Schema Changes

# ❌ BAD — Blocking DDL (might lock table on MySQL, unsupported on Vitess)
# ALTER TABLE orders MODIFY COLUMN status VARCHAR(50) NOT NULL;

# ✅ GOOD — Online DDL using PlanetScale branching + deploy request
# 1. Create a branch: pscale branch create myapp-add-status-index
# 2. Make schema change on the branch
# 3. Create a deploy request
# 4. Deploy (zero-downtime)

---

Constraints

MUST DO

• Always connect with SSL (ssl_ca + ssl_mode=VERIFY_IDENTITY) — PlanetScale rejects non-SSL connections
• Use connection pooling for production — PlanetScale proxy connections have overhead
• Use database branching for all schema changes — never run DDL directly on production branches
• Use deploy requests to merge schema changes — enables review and rollback
• Always call conn.commit() explicitly for write operations
• Use parameterized queries with %s placeholders
• Use cursor(dictionary=True) for readable results

MUST NOT DO

• Never connect without SSL — PlanetScale enforces SSL and will drop non-SSL connections
• Do not create connections per request in serverless functions — use a pool or connection proxy
• Avoid FOREIGN KEY constraints — PlanetScale (Vitess) does not support them
• Never run ALTER TABLE directly on production branches — use branching + deploy requests
• Do not use MyISAM tables — PlanetScale only supports InnoDB
• Avoid subqueries in FROM clauses that are not correlated — Vitess may not optimize them well
• Never hardcode database credentials — use environment variables

---

Output Template

When writing PlanetScale integration code, structure your output as:

1. Connection Setup — mysql.connector.connect() with SSL config, host, credentials
2. Query Execution — Parameterized SQL with cursor(dictionary=True) for dict results
3. Pool Management — MySQLConnectionPool or SQLAlchemy engine for connection reuse
4. Branching / Deploy Request — PlanetScale API calls for branch management
5. Error Handling — Catch MySQLError, handle connection drops, rollback on write failures

---

Related Skills

Skill	Purpose
coding-postgresql-sdk	PostgreSQL SDK patterns (alternative relational database)
coding-supabase-sdk	Supabase BaaS patterns for PostgreSQL
coding-database-migrations	Migration workflow patterns compatible with PlanetScale branching

---

Live References

• PlanetScale Python Connection Guide — Official Python connection examples
• PlanetScale API Reference — REST API for branch and deploy request management
• mysql-connector-python Docs — Official MySQL Connector/Python documentation
• PlanetScale Branching Docs — Database branching workflow
• PlanetScale Deploy Requests — Schema change deploy workflow
• PlanetScale with SQLAlchemy — SQLAlchemy engine configuration
• Vitess MySQL Compatibility — MySQL features supported by PlanetScale