SQLAlchemy dialect integrated into Cloud Spanner database

Warnings

• gotcha The database URL format for `sqlalchemy-spanner` differs between SQLAlchemy 1.3 and 1.4+/2.0. For SQLAlchemy 1.3, use `spanner:///...`. For SQLAlchemy 1.4+ and 2.0, use `spanner+spanner:///...`.
  Fix: Adjust the connection string based on your SQLAlchemy version. E.g., `create_engine("spanner+spanner:///projects/...")` for newer SQLAlchemy versions.
• gotcha Cloud Spanner does not support DDL (Data Definition Language) statements within transactions. Any DDL operations performed will not be rolled back upon transaction failure.
  Fix: Be mindful that DDL statements are implicitly committed and irreversible in the context of a transaction. Consider using `connection.execute(text('DDL_STATEMENT'))` followed by `connection.commit()` outside of explicit transaction blocks for DDL.
• gotcha The dialect and the underlying DB API driver do not support Spanner mutations directly; only DML (Data Manipulation Language) statements are supported for updates.
  Fix: Utilize SQLAlchemy's ORM or Core DML statements (INSERT, UPDATE, DELETE) which are translated into Spanner-compatible DML.
• gotcha When using Alembic for migrations, setting `version_table_pk` to `False` in your Alembic environment configuration is highly recommended to avoid issues with Spanner's primary key restrictions on the `alembic_versions` table.
  Fix: In your `env.py` or Alembic configuration, ensure `version_table_pk = False` for Spanner migrations.
• gotcha The `sqlalchemy-spanner` dialect is currently synchronous only. There is no official asynchronous dialect available, which can be a blocker for projects requiring an async engine with SQLAlchemy 2.0+'s async features.
  Fix: Consider using thread pools or synchronous execution within an async application, or explore third-party async wrappers. A native async SpannerSessionService is a feature request for the underlying `google-cloud-spanner` client.
• gotcha Using `autoload=True` with `MetaData` in a highly concurrent environment can lead to `sqlalchemy.exc.CompileError: Unconsumed column names` due to race conditions during table reflection. It's an anti-pattern to reflect tables on every statement.
  Fix: Reflect all necessary `Table` objects upfront, once, before your program begins processing requests, rather than dynamically reflecting them within hot code paths.
• gotcha The dialect does not currently propagate the `timeout` execution option to the underlying gRPC client, causing all queries to use the default gRPC timeout of 3600 seconds regardless of `execution_options(timeout=...)`.
  Fix: Be aware of the default long timeout. For specific timeout control, you may need to apply it at the underlying `google-cloud-spanner` client level if direct access is available, or await a future update to the dialect.

Install

• pip install sqlalchemy-spanner Install stable version

Imports

• create_engine

  from sqlalchemy import create_engine

  The Spanner dialect is typically loaded via the connection string, not direct import of a dialect class.

Quickstart

This quickstart demonstrates how to establish a connection to a Google Cloud Spanner database, define a table, insert data, and query it using the SQLAlchemy ORM. Ensure you have a GCP project, Spanner instance, and database set up, and that your application has appropriate authentication (e.g., via `GOOGLE_APPLICATION_CREDENTIALS`). The example shows direct DDL for table creation, as DDL is not transactional in Spanner.

import os
from sqlalchemy import create_engine, text, MetaData, Table, Column, String, Integer

# Set these environment variables or replace directly
project_id = os.environ.get('SPANNER_PROJECT_ID', 'your-gcp-project-id')
instance_id = os.environ.get('SPANNER_INSTANCE_ID', 'your-spanner-instance-id')
database_id = os.environ.get('SPANNER_DATABASE_ID', 'your-spanner-database-id')

# For SQLAlchemy 1.4+ and 2.0, use 'spanner+spanner:///'
# For SQLAlchemy 1.3, use 'spanner:///'
db_url = f"spanner+spanner:///projects/{project_id}/instances/{instance_id}/databases/{database_id}"

try:
    engine = create_engine(db_url)
    
    # Example: Define a table for demonstration
    metadata = MetaData()
    users = Table(
        'users',
        metadata,
        Column('user_id', String(36), primary_key=True),
        Column('name', String(255), nullable=False),
        Column('age', Integer)
    )
    
    # Create tables (DDL operations are not transactional in Spanner)
    with engine.connect() as connection:
        connection.execute(text("CREATE TABLE IF NOT EXISTS users (user_id STRING(36) NOT NULL, name STRING(255) NOT NULL, age INT64) PRIMARY KEY (user_id)"))
        connection.commit()
        print("Table 'users' ensured.")

    # Insert data
    with engine.begin() as connection:
        connection.execute(users.insert(), {"user_id": "user1", "name": "Alice", "age": 30})
        connection.execute(users.insert(), {"user_id": "user2", "name": "Bob", "age": 24})
        print("Data inserted.")

    # Query data
    with engine.connect() as connection:
        result = connection.execute(users.select().where(users.c.age > 25)).fetchall()
        print("Users older than 25:", result)

except Exception as e:
    print(f"An error occurred: {e}")
    print("Ensure Cloud Spanner project, instance, and database are correctly configured and `GOOGLE_APPLICATION_CREDENTIALS` is set or default credentials are available.")