Query PostgreSQL from a Python Edge Function

Edge functions are for lightweight API routes — authentication checks, webhook handlers, simple CRUD operations. They boot on demand, run a few lines of code, return a response, and shut down. No Dockerfile, no uvicorn server, no Kubernetes pod management. Just a Python script and a requirements.txt.

The catch: most tutorials show edge functions returning hard-coded JSON. Real apps query databases. Here's how to connect a Python edge function to a managed PostgreSQL instance, handle connection pooling in a serverless environment, and avoid the "too many connections" error that kills naive implementations.

Provisioning the PostgreSQL database

Edge functions don't include a database. You provision one separately and link it via an environment variable.

curl -X POST https://api.pandastack.io/v1/databases \
  -H "Authorization: Bearer $PANDASTACK_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "edge-db",
    "engine": "postgres",
    "version": "16"
  }'

Response: { "success": true, "data": { "databaseId": "db_abc123", "connectionString": "postgresql://..." } }.

Save the connection string. You'll pass it to the edge function as DATABASE_URL.

Writing the edge function

Edge functions expose a handler(event) function. The event dict includes method, path, headers, body, and queryStringParameters. You return a dict with statusCode, headers, and body.

Create function.py:

import os
import json
import psycopg2
from psycopg2.extras import RealDictCursor

def handler(event):
    database_url = os.environ.get('DATABASE_URL')
    if not database_url:
        return {
            'statusCode': 500,
            'body': json.dumps({'error': 'DATABASE_URL not set'})
        }

    try:
        conn = psycopg2.connect(database_url, sslmode='require')
        cursor = conn.cursor(cursor_factory=RealDictCursor)

        cursor.execute('SELECT id, email FROM users LIMIT 10')
        users = cursor.fetchall()

        cursor.close()
        conn.close()

        return {
            'statusCode': 200,
            'headers': {'Content-Type': 'application/json'},
            'body': json.dumps({'users': users})
        }
    except Exception as e:
        return {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }

requirements.txt:

psycopg2-binary

This connects to Postgres, runs a query, and returns JSON. The sslmode='require' parameter is important — managed databases enforce SSL by default. Without it, the connection is rejected.

Deploying the edge function

Edge functions are deployed via the API or CLI. Here's the API route:

curl -X POST https://api.pandastack.io/v1/functions \
  -H "Authorization: Bearer $PANDASTACK_TOKEN" \
  -F "name=get-users" \
  -F "runtime=python" \
  -F "code=@function.py" \
  -F "requirements=@requirements.txt" \
  -F "env[DATABASE_URL]=$DATABASE_URL"

The runtime is python (not python3.11 or python:3.11 — just python). The platform uses Python 3.11 by default.

Response: { "success": true, "data": { "functionId": "fn_xyz", "invokeUrl": "https://functions.pandastack.app/invoke/fn_xyz" } }.

The invoke URL is the live endpoint. Send a GET request:

curl https://functions.pandastack.app/invoke/fn_xyz

Returns:

{
  "users": [
    {"id": 1, "email": "alice@example.com"},
    {"id": 2, "email": "bob@example.com"}
  ]
}

The connection pooling problem

The code above works for low traffic (a few requests per minute). At higher load, you hit "too many connections" errors. Each invocation opens a new Postgres connection, runs the query, and closes the connection. If ten requests arrive simultaneously, you open ten connections. Postgres has a connection limit (50 on the free tier), and you exhaust it quickly.

The fix: connection pooling. But edge functions are stateless — they boot, run once, and shut down. You can't hold a persistent connection pool across invocations.

Two solutions:

1. Use a connection pooler (PgBouncer)

Deploy a PgBouncer instance (a separate service, not part of the edge function) that sits between the function and Postgres. The function connects to PgBouncer, which maintains a pool of connections to Postgres.

This is the production pattern but requires deploying a second service. For hobby projects, the second option is simpler.

2. Reuse connections across invocations (if the runtime supports it)

Some edge function platforms keep the runtime warm for a few seconds between invocations. You can store a connection in a global variable and reuse it:

import os
import json
import psycopg2
from psycopg2.extras import RealDictCursor

# Global connection (reused across invocations if the runtime stays warm)
_conn = None

def get_connection():
    global _conn
    if _conn is None or _conn.closed:
        database_url = os.environ.get('DATABASE_URL')
        _conn = psycopg2.connect(database_url, sslmode='require')
    return _conn

def handler(event):
    try:
        conn = get_connection()
        cursor = conn.cursor(cursor_factory=RealDictCursor)

        cursor.execute('SELECT id, email FROM users LIMIT 10')
        users = cursor.fetchall()

        cursor.close()

        return {
            'statusCode': 200,
            'headers': {'Content-Type': 'application/json'},
            'body': json.dumps({'users': users})
        }
    except Exception as e:
        return {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }

Now the first invocation opens a connection and stores it in _conn. Subsequent invocations reuse it (if the runtime is still warm). If the runtime shuts down and a new one boots, _conn is None again, and a new connection is opened.

This reduces connection churn but doesn't guarantee you won't hit the limit under heavy load. For production, use PgBouncer.

Using the CLI to deploy

panda login
panda functions deploy \
  --name get-users \
  --runtime python \
  --code function.py \
  --requirements requirements.txt \
  --env DATABASE_URL=$DATABASE_URL

The CLI reads the local files and uploads them. After deploy, you get an invoke URL.

Handling POST requests

If the function receives POST data (e.g. creating a user), read event['body']:

def handler(event):
    if event['method'] == 'POST':
        data = json.loads(event['body'])
        email = data.get('email')

        conn = get_connection()
        cursor = conn.cursor()
        cursor.execute('INSERT INTO users (email) VALUES (%s) RETURNING id', (email,))
        user_id = cursor.fetchone()[0]
        conn.commit()
        cursor.close()

        return {
            'statusCode': 201,
            'body': json.dumps({'id': user_id, 'email': email})
        }

Send a POST request:

curl -X POST https://functions.pandastack.app/invoke/fn_xyz \
  -H "Content-Type: application/json" \
  -d '{"email": "charlie@example.com"}'

Returns {"id": 3, "email": "charlie@example.com"}.

Debugging "SSL connection required"

If the function returns an error about SSL, check the connection string. Managed Postgres enforces SSL by default. Add sslmode=require to the DSN:

psycopg2.connect(database_url, sslmode='require')

Or append ?sslmode=require to the connection string if it's not already there.

Comparing Python to Node.js edge functions

PandaStack supports two edge function runtimes: Python and Node.js (no Go or Rust). For database queries:

• Python + psycopg2: mature driver, good connection pooling with psycopg2.pool.
• Node.js + pg: similar maturity, async/await-native, slightly lower cold start time.

If your backend is Python (FastAPI, Flask), use Python functions. If it's Node.js (Express, Hono), use Node.js functions. The runtime is a parameter in the deploy payload (runtime: "python" or runtime: "nodejs").

What you've deployed

A Python edge function that queries a managed PostgreSQL database and returns JSON. The function boots on demand, runs the query, and shuts down. Connection pooling is handled either by reusing a global connection (if the runtime stays warm) or by deploying a PgBouncer instance for production traffic.

Edge functions are billed per invocation and execution time (similar to AWS Lambda). Free-tier apps include a generous allocation of invocations. For apps with steady traffic, a container (FastAPI with Uvicorn) is more cost-effective. For bursty traffic (webhooks, scheduled tasks), edge functions are cheaper because you don't pay for idle time.

PandaStack's edge functions run on a shared OpenWhisk cluster (the platform's serverless runtime). They auto-scale based on load, with cold starts typically under 500ms for Python (faster for Node.js). For sub-100ms cold starts, use a container instead.

The managed PostgreSQL instance runs continuously (no scale-to-zero for databases). Daily backups are automatic, with retention based on your plan (7/15/30 days for Free/Pro/Premium). In-place restore from backup is available on paid plans.

References

• [psycopg2 documentation](https://www.psycopg.org/docs/)
• [PostgreSQL connection strings](https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNSTRING)
• [PandaStack edge functions](https://docs.pandastack.io/functions/)
• [PandaStack databases](https://docs.pandastack.io/databases/)