sqlmodel-gcp-postgres

SQLModel ORM integration for Google Cloud SQL PostgreSQL instances with seamless support for local development and testing.

Features

• Cloud SQL Integration: Uses the official Google Cloud SQL Python Connector for secure, simplified connections
• Multi-Database Support: Works with PostgreSQL, SQLite, and other databases for local development and testing
• Transaction Management: Built-in decorator for automatic commit/rollback handling
• Lazy Connection Refresh: Compatible with Cloud Run and other auto-scaling environments

Installation

# pip
pip install sqlmodel-gcp-postgres
# uv
uv add sqlmodel-gcp-postgres
# Poetry
poetry add sqlmodel-gcp-postgres

Note: This library requires the pg8000 database driver and is designed for use with PostgreSQL and Cloud SQL. It is included as a dependency in the cloud-sql-python-connector[pg8000] requirement.

Quick Start

Cloud SQL (Production)

Set environment variables:

USE_CLOUD_SQL=true
CLOUD_SQL_INSTANCE_CONNECTION_NAME=your-project:region:instance-name
DB_USER=postgres-user
DB_PASS=postgres-password
DB_NAME=database-name

Optional: Use private IP:

CLOUD_SQL_PRIVATE_IP=true

Local Development with PostgreSQL (Docker Compose)

services:
  db:
    image: postgres:17-alpine
    environment:
      POSTGRES_USER: dev_user
      POSTGRES_PASSWORD: dev_password
      POSTGRES_DB: dev_db
    ports:
      - "5432:5432"
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U dev_user"]
      interval: 10s
      timeout: 5s
      retries: 5

  app:
    build: .
    depends_on:
      db:
        condition: service_healthy
    environment:
      DB_URL: postgresql+pg8000://dev_user:dev_password@db:5432/dev_db

Testing with SQLite

import os
os.environ["DB_URL"] = "sqlite:///test.db"

Usage

from sqlmodel import Session, select
from sqlmodel_gcp_postgres import get_session, transactional

# Use in dependency injection (e.g., FastAPI)
@app.post("/api/users")
@transactional
async def create_user(user: User, session: Session = Depends(get_session)):
    session.add(user)
    return user  # Auto-commits on return

# Or with context manager
def query_users():
    session = get_session()
    users = session.exec(select(User)).all()
    return users

Configuration

The library automatically detects the database mode based on USE_CLOUD_SQL:

License

MIT License - see LICENSE file for details.