Connect with Python

Connect to PhoenixDB from your Python application using psycopg2 or SQLAlchemy.

Option 1: psycopg2

Install

pip install psycopg2-binary

Basic Connection

db.py

import os
import psycopg2
from psycopg2.extras import RealDictCursor

def query(sql, params=None):
    conn = psycopg2.connect(
        os.environ['DATABASE_URL'],
        cursor_factory=RealDictCursor
    )
    try:
        with conn.cursor() as cur:
            cur.execute(sql, params)
            if cur.description:
                return cur.fetchall()
            return None
    finally:
        conn.close()

Usage Example

app.py

from db import query

# Create table
query("""
CREATE TABLE IF NOT EXISTS users (
    id SERIAL PRIMARY KEY,
    name TEXT NOT NULL,
    email TEXT UNIQUE NOT NULL
);
""")

# Insert
user = query(
    "INSERT INTO users (name, email) VALUES (%s, %s) RETURNING *",
    ("Jane", "jane@example.com")
)
print(user)

# Select
users = query("SELECT * FROM users")
print(users)

Option 2: SQLAlchemy

Install

pip install sqlalchemy psycopg2-binary

Database Setup

database.py

import os
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

DATABASE_URL = os.environ['DATABASE_URL']

engine = create_engine(
    DATABASE_URL,
    connect_args={"sslmode": "require"}
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Model Definition

models.py

from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.sql import func
from database import Base

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, nullable=False)
    email = Column(String, unique=True, index=True, nullable=False)
    created_at = Column(DateTime(timezone=True), server_default=func.now())

Usage with FastAPI

main.py

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from database import get_db, engine, Base
from models import User

Base.metadata.create_all(bind=engine)
app = FastAPI()

@app.get('/users')
def list_users(db: Session = Depends(get_db)):
    return db.query(User).all()

@app.post("/users")
def create_user(name: str, email: str, db: Session = Depends(get_db)):
    user = User(name=name, email=email)
    db.add(user)
    db.commit()
    db.refresh(user)
    return user

Environment Variable

.env

DATABASE_URL=postgresql://postgres:YOUR_PASSWORD@abc123.server1.phoenixdb.space:5432/mydb?sslmode=require

Load with: pip install python-dotenv andload_dotenv()

Django Configuration

For Django projects, update settings.py:

settings.py

import dj_database_url

DATABASES = {
    'default': dj_database_url.config(
        default=os.environ.get('DATABASE_URL'),
        conn_max_age=600,
        ssl_require=True,
    )
}

Install: pip install dj-database-url

Next Steps

Connection Strings →

Learn about connection options

Import Data →

Migrate existing database