Planetscale Sdk

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

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