Flask

Flask connects to DB9 through SQLAlchemy and a standard PostgreSQL driver. This guide uses psycopg2; the psycopg (v3) driver works as well. DB9 is PostgreSQL-compatible, so no special adapter is needed.

This guide shows how to set up a Flask app with DB9 using Flask-SQLAlchemy for model management and Flask-Migrate for schema migrations.

Prerequisites

A DB9 database (create one)
Python 3.10+
Flask 3.0+

Create a DB9 Database

db9 create --name flask-app

db9 create prints the connection string, with the admin password, once — copy it from the Connection String: line. It has this shape:

postgresql://YOUR_TENANT_ID.admin:YOUR_PASSWORD@pg.d0000.db9.io:5432/postgres

YOUR_TENANT_ID is the tenant ID prefix in the generated connection string. Copy the username from the Connection String: line printed by db9 create or from db9 db connect. The examples below use pg.db9.io:5433, which reaches the same server; either host and port works. db9 db status flask-app shows the connection string without the password; if you lose it, db9 db reset-password flask-app sets a new password and prints a new connection string.

Set the connection string as an environment variable:

DATABASE_URL="postgresql+psycopg2://YOUR_TENANT_ID.admin:YOUR_PASSWORD@pg.db9.io:5433/postgres?sslmode=require"

Project Setup

mkdir flask-db9 && cd flask-db9
python -m venv venv
source venv/bin/activate
pip install flask flask-sqlalchemy flask-migrate psycopg2-binary python-dotenv

Configure the App

import os
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from dotenv import load_dotenv

load_dotenv()

db = SQLAlchemy()
migrate = Migrate()

# DB9's pg_table_is_visible() is true for every table in every schema, so Alembic sees
# pg_cron's tables (and tables in your other schemas) as strays and would try to drop
# them. Compare only the tables your models define.
def include_name(name, type_, parent_names):
    return type_ != "table" or name in db.metadata.tables


def create_app():
    app = Flask(__name__)
    app.config["SQLALCHEMY_DATABASE_URI"] = os.environ["DATABASE_URL"]
    app.config["SQLALCHEMY_ENGINE_OPTIONS"] = {
        "pool_size": 5,
        "pool_recycle": 300,
    }

    db.init_app(app)
    migrate.init_app(app, db, include_name=include_name)

    from app import routes
    app.register_blueprint(routes.bp)

    return app

Define Models

from datetime import datetime, timezone
from app import db


class User(db.Model):
    __tablename__ = "users"

    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String(255), unique=True, nullable=False)
    name = db.Column(db.String(100), nullable=False)
    created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
    posts = db.relationship("Post", backref="author", lazy=True)


class Post(db.Model):
    __tablename__ = "posts"

    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(500), nullable=False)
    content = db.Column(db.Text)
    published = db.Column(db.Boolean, default=False)
    author_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=False)
    created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))

Run Migrations

Run these once app/routes.py from the next section exists — create_app() imports the routes module, so flask db init fails with an ImportError before then.

flask db init
flask db migrate -m "create users and posts"
flask db upgrade

Routes

from flask import Blueprint, jsonify, request
from app import db
from app.models import User

bp = Blueprint("api", __name__, url_prefix="/api")


@bp.route("/users", methods=["GET"])
def list_users():
    users = User.query.order_by(User.created_at.desc()).all()
    return jsonify([
        {"id": u.id, "name": u.name, "email": u.email}
        for u in users
    ])


@bp.route("/users", methods=["POST"])
def create_user():
    data = request.get_json()
    user = User(name=data["name"], email=data["email"])
    db.session.add(user)
    db.session.commit()
    return jsonify({"id": user.id, "name": user.name, "email": user.email}), 201

Run the App

from app import create_app

app = create_app()

flask --app wsgi run --debug

Test with curl:

curl -X POST http://localhost:5000/api/users \
  -H "Content-Type: application/json" \
  -d '{"name": "Alice", "email": "alice@example.com"}'

curl http://localhost:5000/api/users

Production Notes

Server-side only: DB9 connections happen on the server. Never expose the connection string to client-side code.
Connection pooling: Start with pool_size=5. DB9 handles per-tenant connection pooling on the server side.
TLS: Use sslmode=require in the connection string.
Host and port: pg.db9.io:5433 and the pg.d0000.db9.io:5432 that db9 create prints reach the same server; port 5432 works on either host.
WSGI server: Use Gunicorn or uWSGI in production instead of the Flask dev server.

Troubleshooting

`ECONNREFUSED` on port 5432

DB9 accepts port 5432 as well as 5433, so the client is not reaching DB9 at all. A refusal from localhost or 127.0.0.1 means the connection string never reached the driver — check that DATABASE_URL is set in the process that opens the connection.

`psycopg2` not found

Install psycopg2-binary for development or psycopg2 (with libpq) for production:

pip install psycopg2-binary

`No module named 'psycopg'`

Your DATABASE_URL starts with a bare postgresql://, which SQLAlchemy 2.1 resolves to the psycopg (v3) driver. Change the scheme to postgresql+psycopg2:// to use the psycopg2-binary driver installed above.

Migration errors

flask db migrate, flask db upgrade, flask db current and flask db downgrade all work against DB9 with the setup in this guide — the include_name hook in app/__init__.py, on psycopg2 or psycopg (v3): migrate detects new tables and added columns, and reports No changes in schema detected when there are none.

The hook is not optional. DB9’s pg_table_is_visible() returns true for every table in every schema, whatever the search_path — including the three tables of the pre-installed pg_cron extension (job, job_run_details and running_jobs in the cron schema) and any table in a schema of your own. Alembic takes all of them for stray tables in the default schema. Without the hook, every flask db migrate adds lines like these to the generated migration (in no fixed order), and flask db upgrade then fails on the first one without applying anything:

op.drop_table('job_run_details')
op.drop_table('job')
op.drop_table('running_jobs')

sqlalchemy.exc.ProgrammingError: (psycopg2.errors.UndefinedTable) relation "job_run_details" does not exist

If you already generated such a migration, delete those op.drop_table lines (and the matching op.create_table calls in downgrade()), then add the hook.

The hook has one cost: Alembic now looks only at tables your models define, so deleting a model no longer generates a drop_table — flask db migrate reports No changes in schema detected. Write that drop by hand. New tables and added columns are still detected.

It also filters by name only. If another schema holds a table with the same name as one of your models, the two are reflected as one table with the columns and indexes of both, and flask db migrate proposes drop_column and drop_index operations for the other schema’s parts. Avoid the duplicate name, or delete those operations from the generated migration.

SSL connection errors

Ensure sslmode=require is in your connection string. If using psycopg2, the SSL parameters are passed through the connection string automatically.

SQLAlchemy — full SQLAlchemy integration guide
Connect — connection strings and authentication
Production Checklist — deployment readiness