planetscale-sdk

star 4

Integrates PlanetScale using mysql-connector-python with patterns for database branching, deploy requests, schema management, connection pooling, and insights for serverless MySQL.

paulpas By paulpas schedule Updated 6/4/2026
play_arrow Run Skill in Manus View GitHub

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

Install via CLI
npx skills add https://github.com/paulpas/agent-skill-router --skill planetscale-sdk
Repository Details
star Stars 4
call_split Forks 0
navigation Branch main
article Path SKILL.md
More from Creator
paulpas
paulpas Explore all skills →